From 794d429367fbbc80e68db687b39718396d399835 Mon Sep 17 00:00:00 2001
From: timedout <git@nexy7574.co.uk>
Date: Fri, 10 Jul 2026 19:18:59 +0100
Subject: [PATCH] refactor: Reduce complexity of incoming PDU handling

* Ignores any events pushed without a room ID
* Removes needless clones for PDU size checking
* Removes incorrect ACL check in incoming handler
* Fast-path handling already handled outlier events
* Remove redundant same-room checks, replace with useful ones
* Combine event rejection and persistence in a single function
* Additional safety assertions in upgrade task
* Split upgrade task into multiple subroutines for reduced cognitive complexity
* Only mutate current state and forward extremities in tandem
* Improve code documentation to better explain the logic flow
---
 src/api/server/send.rs                        |  19 +-
 .../event_handler/handle_incoming_pdu.rs      | 116 +---
 .../rooms/event_handler/handle_outlier_pdu.rs | 106 +--
 .../event_handler/upgrade_outlier_pdu.rs      | 632 ++++++++++--------
 src/service/rooms/membership/mod.rs           |   2 +-
 src/service/rooms/timeline/create.rs          |   6 +-
 6 files changed, 472 insertions(+), 409 deletions(-)

diff --git a/src/api/server/send.rs b/src/api/server/send.rs
index 9f8ab6a8f..dde08a100 100644
--- a/src/api/server/send.rs
+++ b/src/api/server/send.rs
@@ -152,7 +152,16 @@ async fn process_inbound_transaction(
 		.stream()
 		.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu))
 		.inspect_err(|e| warn!("Could not parse incoming PDU: {e}"))
-		.ready_filter_map(Result::ok);
+		.ready_filter_map(Result::ok)
+		.ready_filter_map(|pdu| {
+			// Room IDs should always be present on events being pushed to us in txns.
+			// This also ensures incoming PDUs have the right room ID.
+			pdu.2
+				.get("room_id")
+				.and_then(|r| r.as_str())
+				.is_some_and(|v| v == pdu.0.as_str())
+				.then_some(pdu)
+		});
 
 	let edus = body
 		.edus
@@ -241,7 +250,7 @@ async fn handle(
 	pdus: impl Stream<Item = Pdu> + Send,
 	edus: impl Stream<Item = Edu> + Send,
 ) -> std::result::Result<ResolvedMap, TransactionError> {
-	// group pdus by room
+	// Group PDUs by room for parallel processing
 	let pdus = pdus
 		.collect()
 		.map(|mut pdus: Vec<_>| {
@@ -252,7 +261,7 @@ async fn handle(
 		})
 		.await;
 
-	// we can evaluate rooms concurrently
+	// Evaluate rooms in parallel
 	let results: ResolvedMap = pdus
 		.into_iter()
 		.try_stream()
@@ -266,7 +275,7 @@ async fn handle(
 		.boxed()
 		.await?;
 
-	// evaluate edus after pdus, at least for now.
+	// Evaluate EDUs after PDUs in case some of the PDUs then forbid some EDUs.
 	edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
 		.boxed()
 		.await;
@@ -322,7 +331,7 @@ async fn handle_room(
 		let result = services
 			.rooms
 			.event_handler
-			.handle_incoming_pdu(origin, room_id, &event_id, value.clone(), true)
+			.handle_incoming_pdu(origin, room_id, &event_id, value, true)
 			.boxed()
 			.await
 			.map(|_| ());
diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs
index b5343d825..2c00c527f 100644
--- a/src/service/rooms/event_handler/handle_incoming_pdu.rs
+++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs
@@ -4,46 +4,16 @@ use std::{
 };
 
 use conduwuit::{
-	Err, Event, Result, debug, debug_error, debug_warn, defer, err, error, info,
-	matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn,
+	Err, Event, Result, debug, debug_error, debug_warn, defer, error, matrix::PartialPdu,
+	result::DebugInspect, trace, utils::time::jitter, warn,
 };
-use futures::{
-	FutureExt, StreamExt,
-	future::{OptionFuture, try_join4},
-};
-use ruma::{CanonicalJsonValue, EventId, OwnedUserId, RoomId, ServerName, UserId};
+use futures::{FutureExt, StreamExt, future::try_join3};
+use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UserId};
 use tokio::sync::mpsc;
 
 use crate::rooms::timeline::{RawPduId, pdu_fits};
 
 impl super::Service {
-	/// When receiving an event one needs to:
-	/// 0. Check the server is in the room
-	/// 1. Skip the PDU if we already know about it
-	/// 1.1. Remove unsigned field
-	/// 2. Check signatures, otherwise drop
-	/// 3. Check content hash, redact if doesn't match
-	/// 4. Fetch any missing auth events doing all checks listed here starting
-	///    at 1. These are not timeline events
-	/// 5. Reject "due to auth events" if can't get all the auth events or some
-	///    of the auth events are also rejected "due to auth events"
-	/// 6. Reject "due to auth events" if the event doesn't pass auth based on
-	///    the auth events
-	/// 7. Persist this event as an outlier
-	/// 8. If not timeline event: stop
-	/// 9. Fetch any missing prev events doing all checks listed here starting
-	///    at 1. These are timeline events
-	/// 10. Fetch missing state and auth chain events by calling `/state_ids` at
-	///     backwards extremities doing all the checks in this list starting at
-	///     1. These are not timeline events
-	/// 11. Check the auth of the event passes based on the state of the event
-	/// 12. Ensure that the state is derived from the previous current state
-	///     (i.e. we calculated by doing state res where one of the inputs was a
-	///     previously trusted set of state, don't just trust a set of state we
-	///     got from a remote)
-	/// 13. Use state resolution to find new room state
-	/// 14. Check if the event passes auth based on the "current state" of the
-	///     room, if not soft fail it
 	#[tracing::instrument(
 		name = "pdu",
 		skip_all,
@@ -57,11 +27,12 @@ impl super::Service {
 		value: BTreeMap<String, CanonicalJsonValue>,
 		is_timeline_event: bool,
 	) -> Result<Option<RawPduId>> {
-		// 1. Skip the PDU if we already have it as a timeline event
+		// Skip the PDU if we already have it as a timeline event. We still re-process
+		// outliers in this scenario.
 		if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await {
 			return Ok(Some(pdu_id));
 		}
-		if !pdu_fits(&mut value.clone()) {
+		if !pdu_fits(&value) {
 			warn!(
 				"dropping incoming PDU {event_id} in room {room_id} from {origin} because it \
 				 exceeds 65535 bytes or is otherwise too large."
@@ -72,34 +43,21 @@ impl super::Service {
 			"processing incoming PDU from {origin} for room {room_id} with event id {event_id}"
 		);
 
-		// 1.1 Check we even know about the room
-		let meta_exists = self.services.metadata.exists(room_id).map(Ok);
+		// If this is a membership state event for a local user, we will be interested
+		// in this event even if we aren't in the room. This allows us to process things
+		// like revoking invites over federation, without being in the room.
+		let is_interesting_member_event = value.get("type").and_then(|t| t.as_str())
+			== Some("m.room.member")
+			&& value
+				.get("state_key")
+				.and_then(|s| s.as_str())
+				.and_then(|s| UserId::parse(s).ok())
+				.is_some_and(|u| self.services.globals.user_is_local(&u));
 
-		// 1.2 Check if the room is disabled
-		let is_disabled = self.services.metadata.is_disabled(room_id).map(Ok);
-
-		// 1.3.1 Check room ACL on origin field/server
-		let origin_acl_check = self.acl_check(origin, room_id);
-
-		// 1.3.2 Check room ACL on sender's server name
-		let sender: OwnedUserId = value
-			.get("sender")
-			.and_then(|v| v.as_str())
-			.ok_or_else(|| err!("No sender in object"))
-			.and_then(|v| Ok(UserId::parse(v)?))
-			.map_err(|e| err!(Request(BadJson("PDU does not have a valid sender key: {e}"))))?;
-
-		let sender_acl_check: OptionFuture<_> = sender
-			.server_name()
-			.ne(origin)
-			.then(|| self.acl_check(sender.server_name(), room_id))
-			.into();
-
-		let (meta_exists, is_disabled, (), ()) = try_join4(
-			meta_exists,
-			is_disabled,
-			origin_acl_check,
-			sender_acl_check.map(|o| o.unwrap_or(Ok(()))),
+		let (room_exists, is_disabled, ()) = try_join3(
+			self.services.metadata.exists(room_id).map(Ok),
+			self.services.metadata.is_disabled(room_id).map(Ok),
+			self.acl_check(origin, room_id),
 		)
 		.await
 		.inspect_err(
@@ -112,21 +70,11 @@ impl super::Service {
 			)));
 		}
 
-		if !self
-			.services
-			.state_cache
-			.server_in_room(self.services.globals.server_name(), room_id)
-			.await
-		{
-			info!(
-				%origin,
-				%room_id,
-				"Dropping inbound PDU for room we aren't participating in"
-			);
-			return Err!(Request(NotFound("This server is not participating in that room.")));
-		}
-
-		if !meta_exists {
+		if !room_exists && !is_interesting_member_event {
+			if is_interesting_member_event {
+				// TODO: handle interesting membership events where we aren't in
+				// the room
+			}
 			return Err!(Request(NotFound("Room is unknown to this server")));
 		}
 
@@ -152,7 +100,7 @@ impl super::Service {
 			.handle_outlier_pdu(origin, create_event, event_id, room_id, value)
 			.await?;
 
-		// 8. if not timeline event: stop
+		// If this is not a timeline event, stop now, as we don't want to de-outlier it.
 		if !is_timeline_event {
 			return Ok(None);
 		}
@@ -169,8 +117,9 @@ impl super::Service {
 			return Ok(None);
 		}
 
-		// 9. Fetch any missing prev events doing all checks listed here starting at 1.
-		//    These are timeline events
+		// Fetch any missing prev events doing all checks listed here starting at 1.
+		// These are timeline events.
+		// TODO: This part needs to be done in a background queue somewhere.
 
 		debug!("Fetching and persisting any missing prev events");
 		Box::pin(self.fetch_prevs(
@@ -185,9 +134,10 @@ impl super::Service {
 			error!("Failed to fetch and persist incoming event's prev_events: {e:?}");
 		})?;
 
-		let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event";
+		let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event"
+			&& incoming_pdu.state_key().is_none();
 
-		// Done with prev events, now handling the incoming event
+		// Done with prev events, now we can handle promoting the PDU
 		let pdu_id = Box::pin(self.upgrade_outlier_to_timeline_pdu(
 			incoming_pdu,
 			val,
diff --git a/src/service/rooms/event_handler/handle_outlier_pdu.rs b/src/service/rooms/event_handler/handle_outlier_pdu.rs
index 72e7b0629..a19f8f857 100644
--- a/src/service/rooms/event_handler/handle_outlier_pdu.rs
+++ b/src/service/rooms/event_handler/handle_outlier_pdu.rs
@@ -16,8 +16,13 @@ use crate::rooms::timeline::pdu_fits;
 impl super::Service {
 	/// Handles a PDU as an outlier, performing basic checks like signatures and
 	/// hashes, proclaimed event auth, and then adding it to the outlier tree.
+	///
+	/// This performs steps 1 through 4 of [S-S section 5.1][spec], returning
+	/// the parsed PDU and modified JSON object.
+	///
+	/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
 	#[allow(clippy::too_many_arguments)]
-	#[tracing::instrument(name="handle_outlier", skip_all, fields(%event_id))]
+	#[tracing::instrument(name = "handle_outlier", skip_all)]
 	pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		&self,
 		origin: &'a ServerName,
@@ -29,54 +34,47 @@ impl super::Service {
 	where
 		Pdu: Event + Send + Sync,
 	{
-		if !pdu_fits(&mut value.clone()) {
+		// Skip outlier handling if we already have this event as either a timeline or
+		// outlier PDU.
+		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));
+		}
+
+		// 1. Check that the PDU follows the format for the room version
+		// (in this case, just size check)
+		if !pdu_fits(&value) {
 			warn!(
 				"dropping incoming PDU {event_id} in room {room_id} from {origin} because it \
 				 exceeds 65535 bytes or is otherwise too large."
 			);
 			return Err!(Request(TooLarge("PDU is too large")));
 		}
-		// 1. Remove unsigned field
-		value.remove("unsigned");
 
-		// 2. Check signatures, otherwise drop
-		// 3. check content hash, redact if doesn't match
+		value.remove("unsigned");
 		let room_version_rules = get_room_version_rules(create_event)?;
+
+		// 2. Check signatures, otherwise drop.
+		// 3. Check content hash, redacting the event if it fails.
 		let mut incoming_pdu = match self
 			.services
 			.server_keys
 			.verify_event(&value, &room_version_rules)
 			.await
 		{
-			| 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::All) => value,
 			| Ok(ruma::signatures::Verified::Signatures) => {
-				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."
-					);
-					check_room_id(room_id, &pdu_event)?;
-					let obj = pdu_event.to_canonical_object();
-					return Ok((pdu_event, obj));
-				}
-
-				debug_info!("Calculated hash does not match (redaction): {event_id}");
+				debug_info!("Content hash mismatch, redacting event and continuing");
 				redact(value, &room_version_rules.redaction, None)
-					.map_err(|e| err!(Request(BadJson("Failed to redact {event_id}: {e}"))))?
+					.map_err(|e| err!(Request(BadJson("Unable to redact {event_id}: {e}"))))?
 			},
 			| Err(e) => {
 				return Err!(Request(Forbidden(debug_error!(
@@ -91,14 +89,11 @@ impl super::Service {
 			"event_id".to_owned(),
 			CanonicalJsonValue::String(event_id.as_str().to_owned()),
 		);
-
 		let pdu_event = serde_json::from_value::<PduEvent>(
 			serde_json::to_value(&incoming_pdu).expect("CanonicalJsonObj is a valid JsonValue"),
 		)
 		.map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?;
 
-		check_room_id(room_id, &pdu_event)?;
-
 		// TODO(nex): From hereon the event is not dropped, and thus always added as an
 		// outlier. However, we only do that at the end of this function, which means
 		// several duplicated calls to add_pdu_outlier. Shouldn't we just do it here
@@ -116,7 +111,6 @@ impl super::Service {
 
 		for auth_event_id in pdu_event.auth_events() {
 			if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await {
-				check_room_id(room_id, &auth_event)?;
 				trace!("Found auth event {auth_event_id} for outlier event {event_id} locally");
 				auth_events.insert(auth_event_id.to_owned(), auth_event);
 			} else {
@@ -154,7 +148,7 @@ impl super::Service {
 		}
 
 		// Ensure none of the auth events are rejected - if they are, reject too.
-		for auth_event_id in auth_events.keys() {
+		for (auth_event_id, auth_event) in &auth_events {
 			if self
 				.services
 				.pdu_metadata
@@ -166,15 +160,27 @@ impl super::Service {
 					 {auth_event_id}",
 					event_id,
 				);
-				self.services.pdu_metadata.mark_event_rejected(event_id);
+				self.reject_and_persist(event_id, &incoming_pdu);
 				return Err!(Request(Forbidden(
 					"Event has rejected auth event: {auth_event_id}"
 				)));
 			}
+			if auth_event.room_id_or_hash() != room_id {
+				debug_warn!(
+					%auth_event_id,
+					auth_event_room_id=%auth_event.room_id_or_hash(),
+					expected_room_id=%room_id,
+					"Rejecting incoming event which depends on an auth event in another room.",
+				);
+				self.reject_and_persist(event_id, &incoming_pdu);
+				return Err!(Request(Forbidden(
+					"Event depends on a cross-room auth event: {auth_event_id}"
+				)));
+			}
 		}
 
-		// 6. Reject "due to auth events" if the event doesn't pass auth based on the
-		//    auth events
+		// 4. Reject "due to auth events" if the event doesn't pass auth based on the
+		//    claimed auth events
 		debug!("Checking based on auth events");
 		let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len());
 		// Build map of auth events
@@ -184,8 +190,6 @@ impl super::Service {
 				.expect("we just checked that we have all auth events")
 				.to_owned();
 
-			check_room_id(room_id, &auth_event)?;
-
 			let key = auth_event.kind().with_state_key(
 				auth_event
 					.state_key
@@ -197,10 +201,7 @@ impl super::Service {
 					v.insert(auth_event);
 				},
 				| hash_map::Entry::Occupied(_) => {
-					self.services
-						.outlier
-						.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
-					self.services.pdu_metadata.mark_event_rejected(event_id);
+					self.reject_and_persist(event_id, &incoming_pdu);
 					return Err!(Request(Forbidden(
 						"Auth event's type and state_key combination exists multiple times: {}, \
 						 {}",
@@ -220,10 +221,7 @@ impl super::Service {
 			)
 			.await
 		{
-			self.services.pdu_metadata.mark_event_rejected(event_id);
-			self.services
-				.outlier
-				.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
+			self.reject_and_persist(event_id, &incoming_pdu);
 			return Err!(Request(Forbidden(
 				"Event authorisation fails based on event's claimed auth events"
 			)));
@@ -291,4 +289,10 @@ impl super::Service {
 		.await
 		.map_err(|e| err!("Event self-authentication failed: {e:?}"))
 	}
+
+	/// Marks the event as rejected and then saves it as an outlier.
+	pub(super) fn reject_and_persist(&self, event_id: &EventId, pdu: &CanonicalJsonObject) {
+		self.services.pdu_metadata.mark_event_rejected(event_id);
+		self.services.outlier.add_pdu_outlier(event_id, pdu);
+	}
 }
diff --git a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs
index c6099ccf8..00a6a72ae 100644
--- a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs
+++ b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs
@@ -1,19 +1,20 @@
-use std::{borrow::Borrow, sync::Arc, time::Instant};
+use std::{borrow::Borrow, collections::HashMap, sync::Arc, time::Instant};
 
 use conduwuit::{
-	Err, Result, debug, debug_error, debug_info, err, info, is_equal_to,
+	Err, Result, debug, debug_error, debug_info, err, info, is_equal_to, is_true,
 	matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
 	result::DebugInspect,
 	trace,
 	utils::{
 		IterStream,
+		mutex_map::Guard,
 		stream::{BroadbandExt, ReadyExt},
 	},
-	warn,
 };
 use futures::{FutureExt, StreamExt, future::ready};
 use ruma::{
-	CanonicalJsonObject, RoomId, ServerName, api::error::ErrorKind, events::StateEventType,
+	CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId, ServerName, api::error::ErrorKind,
+	events::StateEventType, room_version_rules::RoomVersionRules,
 };
 use tokio::join;
 
@@ -48,14 +49,22 @@ impl super::Service {
 		} else if rejected {
 			return Err!(Request(Forbidden("Event has been rejected")));
 		} else if soft_failed {
+			// Soft-failed events cannot be promoted.
 			return Err!(Request(Forbidden("Event has been soft-failed")));
 		}
 
+		// These should never happen, but they're good last-minute sanity checks to
+		// ensure we never promote totally illegal events.
 		assert_eq!(
 			*create_event.kind(),
 			StateEventType::RoomCreate.into(),
 			"tried to upgrade a PDU with a create_event that is not a room create event"
 		);
+		assert_eq!(
+			incoming_pdu.room_id_or_hash(),
+			*room_id,
+			"room ID mismatch: PDU room ID differs from parameter"
+		);
 
 		debug!(
 			event_id = %incoming_pdu.event_id,
@@ -65,266 +74,92 @@ impl super::Service {
 		let min_depth = self.services.metadata.get_mindepth(room_id).await;
 		let room_version_rules = get_room_version_rules(create_event)?;
 
-		// 10. Fetch missing state and auth chain events by calling /state_ids at
-		//     backwards extremities doing all the checks in this list starting at 1.
-		//     These are not timeline events.
+		// We now need to resolve the state before the event so that we can perform PDU
+		// check 5 (event auth passes based on state before the event). To do this, we
+		// either need to have all the prev events locally, or ask a remote server
+		// for the state at the event.
+		let (passes_state_before, state_before) = self
+			.is_authorised_by_state_before(
+				&incoming_pdu,
+				&room_version_rules,
+				create_event,
+				origin,
+			)
+			.await?;
 
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Resolving state at event"
-		);
-		let state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
-			self.state_at_incoming_degree_one(&incoming_pdu).await?
-		} else {
-			self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_rules)
-				.await?
-		};
-		let state_at_incoming_event = match state_at_incoming_event {
-			| Some(s) => s,
-			| None => {
-				trace!("Could not calculate incoming state, asking remote {origin} for it");
-				self.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
-					.await
-					.debug_inspect_err(|e| {
-						debug_error!("Could not fetch state from {origin}: {e}");
-					})?
-			},
-		};
-
-		if state_at_incoming_event.is_empty()
-			&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
-		{
-			// This can happen if the remote sends an event but cannot be reached to fetch
-			// the state at it, and all other servers in the room (which might just be the
-			// unreachable server) are unable to provide required info.
-			// returning an error here allows the upgrade to be attempted at another time.
-			return Err!(Request(Forbidden("Could not resolve incoming state at event")));
-		}
-		trace!(state_events = state_at_incoming_event.len(), "Calculated incoming state");
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Performing auth check to upgrade"
-		);
-		// 11. Check the auth of the event passes based on the state of the event
-		let state_fetch_state = &state_at_incoming_event;
-		let state_fetch = |k: StateEventType, s: StateKey| async move {
-			let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
-
-			let event_id = state_fetch_state.get(&shortstatekey)?;
-			self.services.timeline.get_pdu(event_id).await.ok()
-		};
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Running initial auth check"
-		);
-		// PDU check: 5
-		let auth_check = state_res::event_auth::auth_check(
-			&room_version_rules,
-			&incoming_pdu,
-			None, // TODO: third party invite
-			|ty, sk| state_fetch(ty.clone(), sk.into()),
-			create_event.as_pdu(),
-		)
-		.await
-		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
-
-		if !auth_check {
-			self.services
-				.pdu_metadata
-				.mark_event_rejected(incoming_pdu.event_id());
+		if !passes_state_before {
+			self.reject_and_persist(incoming_pdu.event_id(), &val);
 			return Err!(Request(Forbidden(
 				"Event authorisation fails based on the state before the event"
 			)));
 		}
 
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Gathering auth events"
-		);
-		let auth_events = self
-			.services
-			.state
-			.get_auth_events(
-				room_id,
-				incoming_pdu.kind(),
-				incoming_pdu.sender(),
-				incoming_pdu.state_key(),
-				incoming_pdu.content(),
-				&room_version_rules,
-			)
-			.await?;
+		// Now that we know the event passes both self-authentication, and
+		// authentication based on the state before the event, we need to check that it
+		// passes based on the *current* room state (state across all forward
+		// extremities). If it doesn't, we accept it, but soft-fail it, and this
+		// prevents it being promoted.
 
-		let state_fetch = |k: &StateEventType, s: &str| {
-			let key = k.with_state_key(s);
-			ready(auth_events.get(&key).map(ToOwned::to_owned))
-		};
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Running auth check with claimed state auth"
-		);
-		// PDU check: 6
-		let auth_check = state_res::event_auth::auth_check(
-			&room_version_rules,
-			&incoming_pdu,
-			None, // third-party invite
-			state_fetch,
-			create_event.as_pdu(),
-		)
-		.await
-		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
-		if !auth_check {
-			warn!(
-				event_id = %incoming_pdu.event_id,
-				"Event authorization fails based on the current state of the room"
-			);
-		}
-
-		// Soft fail check before doing state res
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Performing soft-fail check"
-		);
-		let mut soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_rules)) {
-			| (false, _) => true,
-			| (true, None) => false,
-			| (true, Some(redact_id)) => {
-				if !self
-					.services
-					.state_accessor
-					.user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true)
-					.await?
-				{
-					warn!(redacts = %redact_id, "User is not allowed to redact event");
-					true
-				} else {
-					false
-				}
-			},
-		};
-
-		// 13. Use state resolution to find new room state
-		// We start looking at current room state now, so lets lock the room
+		// We lock the room here to prevent the current state from changing beneath us
+		// mid-check.
 		trace!(
 			room_id = %room_id,
 			"Locking the room"
 		);
 		let state_lock = self.services.state.mutex.lock(room_id).await;
+		let passes_current_state = self
+			.is_authorised_by_current_state(&incoming_pdu, &room_version_rules, create_event)
+			.await?;
 
-		let state_ids_compressed: Arc<CompressedState> = self
-			.services
-			.state_compressor
-			.compress_state_events(
-				state_at_incoming_event
-					.iter()
-					.map(|(ssk, eid)| (ssk, eid.borrow())),
-			)
-			.collect()
-			.map(Arc::new)
-			.await;
-
-		if incoming_pdu.state_key().is_some() {
-			debug!("Event is a state-event. Deriving new room state");
-
-			// We also add state after incoming event to the fork states
-			let mut state_after = state_at_incoming_event.clone();
-			if let Some(state_key) = incoming_pdu.state_key() {
-				let shortstatekey = self
+		// Determine whether this PDU should be soft-failed.
+		// If the auth check failed, invariably yes. Otherwise, only if the user isn't
+		// allowed to redact the target event (if any).
+		let mut should_soft_fail =
+			match (passes_current_state, incoming_pdu.redacts_id(&room_version_rules)) {
+				| (false, _) => true,
+				| (true, None) => false,
+				| (true, Some(redact_id)) => self
 					.services
-					.short
-					.get_or_create_shortstatekey(
-						&incoming_pdu.kind().to_string().into(),
-						state_key,
-					)
-					.await;
+					.state_accessor
+					.user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true)
+					.await
+					.is_ok_and(is_true!()),
+			};
 
-				let event_id = incoming_pdu.event_id();
-				state_after.insert(shortstatekey, event_id.to_owned());
-			}
-
-			let new_room_state = self
-				.resolve_state(room_id, &room_version_rules, state_after)
-				.await?;
-
-			// Set the new room state to the resolved state
-			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?;
-		}
-
-		if !soft_fail {
-			// Don't call the below checks on events that have already soft-failed, there's
-			// no reason to re-calculate that.
-			// 14-pre. ask the policy server to sign the event, if possible
+		if !should_soft_fail {
+			// Now we can perform check 7, which is ensuring the event passes policy server
+			// checks.
+			// We explicitly only do this if we aren't already going to soft-fail the event,
+			// since the policy server refusing this event also soft-fails it.
 			debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
-			let tmp_evt_id = val.remove("event_id");
-			if let Err(e) = self
-				.policy_server_allows_event(
-					&incoming_pdu,
-					&mut val,
-					room_id,
-					&room_version_rules,
-					true,
-				)
-				.await
-			{
-				if matches!(e.kind(), ErrorKind::Forbidden) {
-					info!(
-						event_id = %incoming_pdu.event_id,
-						error = %e,
-						"Event has been marked as spam by policy server: {}",
-						e.message(),
-					);
-					soft_fail = true;
-				} else {
-					return Err(e);
-				}
-			} else {
-				debug!(
-					event_id = %incoming_pdu.event_id,
-					"Event has passed policy server check."
-				);
-			}
-			if let Some(id) = tmp_evt_id {
-				val.insert("event_id".to_owned(), id);
-			}
+			should_soft_fail = self
+				.policy_server_check(&incoming_pdu, &mut val, &room_version_rules)
+				.await?;
 
-			// Additionally, if this is a redaction for a soft-failed event, we soft-fail it
-			// also.
-
-			// TODO: this is supposed to hide redactions from policy servers, however, for
-			// full efficacy it also needs to hide redactions for unknown events. This
-			// needs to be investigated at a later time.
+			// TODO: this is supposed to hide redactions from policy servers and janitorial
+			// bots, however, for full efficacy it also needs to hide redactions for
+			// unknown events. This needs to be investigated at a later time.
 			if let Some(redact_id) = incoming_pdu.redacts_id(&room_version_rules) {
 				debug!(
 					redact_id = %redact_id,
-					"Checking if redaction is for a soft-failed event"
+					"Checking if redaction is for a soft-failed/rejected event"
 				);
-				if self
+				if !self
 					.services
 					.pdu_metadata
-					.is_event_soft_failed(&redact_id)
+					.is_event_accepted(&redact_id)
 					.await
 				{
-					info!(
-						redact_id = %redact_id,
-						"Redaction is for a soft-failed event"
+					debug_info!(
+						"Soft-failing valid redaction because it targets a non-accepted event"
 					);
-					soft_fail = true;
+					should_soft_fail = true;
 				}
 			}
 		}
 
+		// 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
@@ -332,40 +167,30 @@ impl super::Service {
 			.get_forward_extremities(room_id)
 			.collect()
 			.await;
-		if !soft_fail {
+		if !should_soft_fail {
 			// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
 			// are not added as forward extremities.
-
-			// Now we calculate the set of extremities this room has after the incoming
-			// event has been applied. We start with the previous extremities (aka leaves)
-			trace!("Calculating extremities");
-			extremities = 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(room_id, &event_id)
-						.await
-						.eq(&false)
-						.then_some(event_id)
-				})
-				.collect::<Vec<_>>()
-				.await;
-			extremities.push(incoming_pdu.event_id().to_owned());
-			debug!(
-				"Retained {} extremities checked against {} prev_events",
-				extremities.len(),
-				incoming_pdu.prev_events().count()
-			);
-			assert!(!extremities.is_empty(), "extremities must not empty");
+			// 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
@@ -374,17 +199,18 @@ impl super::Service {
 				val,
 				extremities.iter().map(Borrow::borrow),
 				state_ids_compressed,
-				soft_fail,
+				should_soft_fail,
 				&state_lock,
 				room_id,
 			)
 			.await?;
-		if soft_fail {
+
+		if should_soft_fail {
 			self.services
 				.pdu_metadata
 				.mark_event_soft_failed(incoming_pdu.event_id());
 
-			info!(
+			debug_info!(
 				elapsed = ?timer.elapsed(),
 				event_id = %incoming_pdu.event_id,
 				"Event was soft failed"
@@ -407,4 +233,276 @@ impl super::Service {
 
 		Ok(pdu_id)
 	}
+
+	/// Checks that the event passes PDU check 5, which ensures that the event
+	/// is authorised based on the state before the event (which is the resolved
+	/// state across all prev events).
+	///
+	/// Returns a boolean indicating whether the event is authorised, and also
+	/// the resolved state before the event for later use. Returns an error if
+	/// state fetching or auth checking fails.
+	async fn is_authorised_by_state_before(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+		origin: &ServerName,
+	) -> Result<(bool, HashMap<u64, OwnedEventId>)> {
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Resolving state at event"
+		);
+		let room_id = incoming_pdu.room_id_or_hash();
+
+		// If the incoming event only has one prev event, we can just use the state at
+		// that event, but otherwise we have to resolve across each fork. If we're
+		// missing even one of the prev events, we have to ask a remote server for help.
+		//
+		// TODO: this can be optimised by only loading auth chain events into memory,
+		// rather than the entire state.
+		let state_before = if incoming_pdu.prev_events().count() == 1 {
+			self.state_at_incoming_degree_one(&incoming_pdu).await?
+		} else {
+			self.state_at_incoming_resolved(&incoming_pdu, &room_id, room_version_rules)
+				.await?
+		};
+		let state_before = match state_before {
+			| Some(s) => s,
+			| None => {
+				trace!("Could not calculate incoming state, asking remote {origin} for it");
+				self.fetch_state(origin, create_event, &room_id, incoming_pdu.event_id())
+					.await
+					.debug_inspect_err(|e| {
+						debug_error!("Could not fetch state from {origin}: {e}");
+					})?
+			},
+		};
+
+		if state_before.is_empty()
+			&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
+		{
+			// This can happen if the remote sends an event but cannot be reached to fetch
+			// the state at it, and all other servers in the room (which might just be the
+			// unreachable server) are unable to provide required info.
+			// returning an error here allows the upgrade to be attempted at another time.
+			return Err!(Request(Forbidden("Could not resolve incoming state before event")));
+		}
+		trace!(state_events = state_before.len(), "Calculated incoming state");
+
+		let state_fetch_state = &state_before;
+		let state_fetch = |k: StateEventType, s: StateKey| async move {
+			let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
+
+			let event_id = state_fetch_state.get(&shortstatekey)?;
+			self.services.timeline.get_pdu(event_id).await.ok()
+		};
+
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Running state-before auth check"
+		);
+
+		// PDU check: 5
+		let auth_check = state_res::event_auth::auth_check(
+			room_version_rules,
+			incoming_pdu,
+			None, // TODO: third party invite
+			|ty, sk| state_fetch(ty.clone(), sk.into()),
+			create_event.as_pdu(),
+		)
+		.await
+		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
+		Ok((auth_check, state_before))
+	}
+
+	/// Checks that the event passes PDU check 6, which ensures that the event
+	/// is authorised based on the room's current state (which is the resolved
+	/// state across all current forward extremities).
+	///
+	/// Returns a boolean indicating whether the event is authorised, or an
+	/// error if the auth check fails.
+	async fn is_authorised_by_current_state(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+	) -> Result<bool> {
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Gathering auth events"
+		);
+		let auth_events = self
+			.services
+			.state
+			.get_auth_events(
+				&incoming_pdu.room_id_or_hash(),
+				incoming_pdu.kind(),
+				incoming_pdu.sender(),
+				incoming_pdu.state_key(),
+				incoming_pdu.content(),
+				room_version_rules,
+			)
+			.await?;
+
+		let state_fetch = |k: &StateEventType, s: &str| {
+			let key = k.with_state_key(s);
+			ready(auth_events.get(&key).map(ToOwned::to_owned))
+		};
+
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Running current state auth check"
+		);
+		state_res::event_auth::auth_check(
+			room_version_rules,
+			incoming_pdu,
+			None, // third-party invite
+			state_fetch,
+			create_event.as_pdu(),
+		)
+		.await
+		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))
+	}
+
+	/// Performs PDU check 7 - does the policy server allow this event.
+	///
+	/// If the policy server forbids the event, false is returned. If there is a
+	/// problem contacting the policy server, or it returns an unrecognised
+	/// response, an appropriate error is returned.
+	async fn policy_server_check(
+		&self,
+		incoming_pdu: &PduEvent,
+		pdu_json: &mut CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<bool> {
+		let event_id = pdu_json
+			.remove("event_id")
+			.expect("event_id should be present in pdu_json at this stage");
+		if let Err(e) = self
+			.policy_server_allows_event(
+				incoming_pdu,
+				pdu_json,
+				&incoming_pdu.room_id_or_hash(),
+				room_version_rules,
+				true,
+			)
+			.await
+			.debug_inspect(|()| {
+				debug!(
+					event_id = %incoming_pdu.event_id,
+					"Event has passed policy server check."
+				);
+			}) {
+			return if matches!(e.kind(), ErrorKind::Forbidden) {
+				info!(
+					event_id = %incoming_pdu.event_id,
+					error = %e,
+					"Event has been marked as spam by policy server: {}",
+					e.message(),
+				);
+				Ok(false)
+			} else {
+				Err(e)
+			};
+		}
+		pdu_json.insert("event_id".to_owned(), event_id);
+		Ok(true)
+	}
+
+	/// 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
+	}
 }
diff --git a/src/service/rooms/membership/mod.rs b/src/service/rooms/membership/mod.rs
index 02e0a8d26..22758ad1d 100644
--- a/src/service/rooms/membership/mod.rs
+++ b/src/service/rooms/membership/mod.rs
@@ -520,7 +520,7 @@ impl Service {
 						return state;
 					},
 				};
-				if !pdu_fits(&mut value.clone()) {
+				if !pdu_fits(&value.clone()) {
 					warn!(
 						"dropping incoming PDU {event_id} in room {room_id} from room join \
 						 because it exceeds 65535 bytes or is otherwise too large."
diff --git a/src/service/rooms/timeline/create.rs b/src/service/rooms/timeline/create.rs
index 9cdd3ea34..2f05ca8cc 100644
--- a/src/service/rooms/timeline/create.rs
+++ b/src/service/rooms/timeline/create.rs
@@ -22,7 +22,9 @@ use serde_json::value::{RawValue, to_raw_value};
 
 use super::RoomMutexGuard;
 
-pub fn pdu_fits(owned_obj: &mut CanonicalJsonObject) -> bool {
+/// Ensures the given PDU fits inside the size limits for a PDU.
+#[must_use]
+pub fn pdu_fits(owned_obj: &CanonicalJsonObject) -> bool {
 	// room IDs, event IDs, senders, types, and state keys must all be <= 255 bytes
 	if let Some(CanonicalJsonValue::String(room_id)) = owned_obj.get("room_id") {
 		if room_id.len() > 255 {
@@ -311,7 +313,7 @@ impl super::Service {
 		pdu_json
 			.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
 		// Verify that the *full* PDU isn't over 64KiB.
-		if !pdu_fits(&mut pdu_json.clone()) {
+		if !pdu_fits(&pdu_json.clone()) {
 			// feckin huge PDU mate
 			return Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")));
 		}