Ensure limits for prev and auth events are respected.

Signed-off-by: Jason Volk <jason@zemos.net>

src/service/rooms/event_handler/fetch_auth.rs

@@ -11,7 +11,7 @@ use ruma::{
 };
 use tuwunel_core::{
 	debug, debug_error, debug_warn, implement,
-	matrix::{PduEvent, event::gen_event_id_canonical_json},
+	matrix::{PduEvent, event::gen_event_id_canonical_json, pdu::MAX_AUTH_EVENTS},
 	trace,
 	utils::stream::{BroadbandExt, IterStream, ReadyExt},
 	warn,
@@ -181,23 +181,16 @@ async fn fetch_auth_chain(
 			);
 		}
 
-		if let Some(auth_events) = value
+		value
 			.get("auth_events")
 			.and_then(CanonicalJsonValue::as_array)
-		{
-			for auth_event in auth_events {
-				match serde_json::from_value::<OwnedEventId>(auth_event.clone().into()) {
-					| Ok(auth_event) => {
-						todo_auth_events.push_back(auth_event);
-					},
-					| _ => {
-						warn!("Auth event id is not valid");
-					},
-				}
-			}
-		} else {
-			warn!("Auth event list invalid");
-		}
+			.into_iter()
+			.flatten()
+			.filter_map(|auth_event| auth_event.try_into().ok())
+			.take(MAX_AUTH_EVENTS)
+			.for_each(|auth_event: &EventId| {
+				todo_auth_events.push_back(auth_event.to_owned());
+			});
 
 		events_in_reverse_order.push((next_id.clone(), value));
 		events_all.insert(next_id);

src/service/rooms/event_handler/fetch_prev.rs

@@ -7,7 +7,7 @@ use ruma::{
 };
 use tuwunel_core::{
 	Result, debug_warn, err, implement,
-	matrix::{Event, PduEvent},
+	matrix::{Event, PduEvent, pdu::MAX_PREV_EVENTS},
 	utils::stream::IterStream,
 };
 
@@ -91,6 +91,11 @@ where
 
 		if pdu.origin_server_ts() > first_ts_in_room {
 			amount = amount.saturating_add(1);
+			debug_assert!(
+				pdu.prev_events().count() <= MAX_PREV_EVENTS,
+				"PduEvent {prev_event_id} has too many prev_events"
+			);
+
 			for prev_prev in pdu.prev_events() {
 				if graph.contains_key(prev_prev) {
 					continue;
@@ -136,5 +141,16 @@ where
 		.await
 		.map_err(|e| err!(Database(error!("Error sorting prev events: {e}"))))?;
 
+	debug_assert_eq!(
+		sorted.len(),
+		graph.len(),
+		"topological sort returned a different number of outputs than inputs"
+	);
+
+	debug_assert!(
+		sorted.len() >= eventid_info.len(),
+		"returned topologically sorted events differ from eventid_info"
+	);
+
 	Ok((sorted, eventid_info))
 }

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -110,7 +110,7 @@ pub async fn handle_incoming_pdu<'a>(
 
 	let room_version = room_version::from_create_event(create_event)?;
 
-	let (incoming_pdu, val) = self
+	let (incoming_pdu, pdu) = self
 		.handle_outlier_pdu(origin, room_id, event_id, pdu, &room_version, false)
 		.await?;
 
@@ -183,7 +183,7 @@ pub async fn handle_incoming_pdu<'a>(
 		origin,
 		room_id,
 		incoming_pdu,
-		val,
+		pdu,
 		&room_version,
 		create_event.event_id(),
 	)