Attempts to tackle #1844

@@ -102,0 +160,4 @@
+		// that many events per-request. Synapse returns 20, and conduwuit+ return 50.
+		// This means with a hard iteration limit, we might give up too early, before
+		// we get a chance to even come close to max_fetch_prev_events. As such, we'll
+		// calculate the max limit based on that config option based on these averages.

@@ -209,0 +350,4 @@
+		Ok(discovered)
+	}
+
+	async fn fetch_and_handle_missing_event_via(

@@ -110,0 +88,4 @@
+		for (i, event_id) in to_persist.iter().enumerate() {
+			debug!(
+				elapsed=?start.elapsed(),
+				"[TODO] Persisting fetched prev event: {event_id} ({}/{})",

@@ -55,0 +28,4 @@
+	/// them directly.
+	///
+	/// The end result is a result containing a map of shortstatekeys to event
+	/// IDs. The underlying `Option` is always `Some`.

@@ -86,0 +282,4 @@
+	/// state events as outliers, for use later.
+	///
+	/// Any events that cannot be persisted are dropped with a warning.
+	/// TODO: make it noisy?

@@ -149,0 +164,4 @@
+				)));
+			}
+		}
+		// TODO: do events received from auth chain need persisting? that sounds

@@ -395,0 +397,4 @@
+		// sometimes end up sending PDUs to us that we aren't yet ready to accept, and
+		// consequently drop. Holding the mutex over the room while processing mitigates
+		// this.
+		let _room_lock = self

@@ -16,3 +27,2 @@
 
-/// Find the event and auth it. Once the event is validated (steps 1 - 8)
-/// it is appended to the outliers Tree.
+const GET_MISSING_EVENTS_MAX_BATCH_SIZE: u16 = 50; // matches src/server/get_missing_events.rs#LIMIT_MAX

@@ -71,0 +78,4 @@
+			.copied()
+			.unwrap_or_else(|| int!(0))
+			.to_string()
+			.parse::<u64>()

@@ -293,0 +248,4 @@
+	debug!("Fetching and persisting any missing prev events");
+	self.fetch_prevs(room_id, create_event, &incoming_pdu, origin)
+		.await
+		.debug_inspect_err(|e| {

@@ -298,0 +278,4 @@
+			.await;
+		let mut local_users = self.services.state_cache.local_users_in_room(room_id);
+		while let Some(user_id) = local_users.next().await {
+			if power_levels.for_user(&user_id)

@@ -298,0 +298,4 @@
+					&state_lock,
+				)
+				.await
+				.inspect(|_| debug!(sender=%user_id, "Successfully sent a dummy event"))

@@ -23,3 +25,3 @@
 	room_id: &'a RoomId,
 	mut value: CanonicalJsonObject,
-	auth_events_known: bool,
+	_auth_events_known: bool,

@@ -133,0 +151,4 @@
+				.map_err(|e| err!(Request(BadJson("Invalid PDU {auth_event_id}: {e}"))))?;
+			auth_chain_map.insert(auth_event_id, auth_pdu);
+		}
+		for aid in pdu_event.auth_events() {

@@ -298,0 +264,4 @@
+		.count()
+		.await;
+
+	if extremities_count >= self.services.server.config.dummy_event_threshold.into() {

@@ -236,3 +231,3 @@
 	}
 
-	// Skip old events
+	// Skip events sent before we joined (they need to be persisted as backfilled

@@ -298,0 +280,4 @@
+		let squash_timings = self.last_extremity_squash.read();
+		squash_timings.get(room_id).copied()
+	};
+	if last_squash.is_some_and(|s| s.elapsed() < Duration::from_mins(1)) {

@@ -56,6 +56,7 @@ impl crate::Service for Service {
 		Ok(Arc::new(Self {
 			mutex_federation: RoomMutexMap::new(),
 			federation_handletime: HandleTimeMap::new().into(),
+			last_extremity_squash: SyncRwLock::new(HashMap::new()),

@@ -312,0 +338,4 @@
+
+			let mut closing = false;
+
+			let waker = tokio::time::sleep(Duration::from_mins(2));

@@ -92,2 +95,4 @@
 
 	fn interrupt(&self) { self.server_shutdown.notify_waiters(); }
+
+	async fn clear_cache(&self) {}