From 46aefa8307cc1d204c5ae36e0425c06d13cbfe5f Mon Sep 17 00:00:00 2001
From: timedout <git@nexy7574.co.uk>
Date: Tue, 23 Jun 2026 15:26:28 +0100
Subject: [PATCH] perf: Throttle dummy events to prevent stampeding

---
 .../event_handler/handle_incoming_pdu.rs      | 100 ++++++++++--------
 src/service/rooms/event_handler/mod.rs        |   7 ++
 2 files changed, 65 insertions(+), 42 deletions(-)

diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs
index 1e37c0355..c08e4bb61 100644
--- a/src/service/rooms/event_handler/handle_incoming_pdu.rs
+++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs
@@ -1,4 +1,7 @@
-use std::{collections::BTreeMap, time::Instant};
+use std::{
+	collections::BTreeMap,
+	time::{Duration, Instant},
+};
 
 use conduwuit::{
 	Err, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, defer, err, error,
@@ -271,49 +274,62 @@ impl super::Service {
 			.await;
 
 		if extremities_count >= self.services.server.config.dummy_event_threshold.into() {
-			debug_warn!(
-				count=%extremities_count,
-				threshold=%self.services.server.config.dummy_event_threshold,
-				"Attempting to squash extremities after upgrading pdu"
-			);
-			// Try to send a dummy event to squash extremities. See issue #1844
-			let power_levels = self
-				.services
-				.state_accessor
-				.get_room_power_levels(room_id)
-				.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.user_can_send_message(&user_id, "org.matrix.dummy_event".into())
-				{
-					trace!(%user_id, "user does not have power level to send dummy event, skipping");
-					continue;
-				}
-				let state_lock = self.services.state.mutex.lock(room_id).await;
-				if self
-					.services
-					.timeline
-					.build_and_append_pdu(
-						PartialPdu {
-							event_type: "org.matrix.dummy_event".into(),
-							..PartialPdu::default()
-						},
-						&user_id,
-						Some(room_id),
-						&state_lock,
-					)
-					.await
-					.inspect(|_| debug!(sender=%user_id, "Successfully sent a dummy event"))
-					.inspect_err(
-						|e| debug!(sender=%user_id, ?e, "Failed to send a dummy event via user"),
-					)
-					.is_ok()
-				{
-					break;
-				}
-			}
+			self.squash_extremities(room_id, extremities_count).await;
 		}
 
 		Ok(pdu_id)
 	}
+
+	async fn squash_extremities(&self, room_id: &RoomId, count: usize) {
+		let last_squash = {
+			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)) {
+			// Avoid sending more than one squash per minute to avoid flooding rooms.
+			return;
+		}
+		debug_warn!(
+			%count,
+			threshold=%self.services.server.config.dummy_event_threshold,
+			"Attempting to squash extremities after upgrading pdu"
+		);
+		// Try to send a dummy event to squash extremities. See issue #1844
+		let power_levels = self
+			.services
+			.state_accessor
+			.get_room_power_levels(room_id)
+			.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.user_can_send_message(&user_id, "org.matrix.dummy_event".into()) {
+				trace!(%user_id, "user does not have power level to send dummy event, skipping");
+				continue;
+			}
+			let state_lock = self.services.state.mutex.lock(room_id).await;
+			if self
+				.services
+				.timeline
+				.build_and_append_pdu(
+					PartialPdu {
+						event_type: "org.matrix.dummy_event".into(),
+						..PartialPdu::default()
+					},
+					&user_id,
+					Some(room_id),
+					&state_lock,
+				)
+				.await
+				.inspect(|_| debug!(sender=%user_id, "Successfully sent a dummy event"))
+				.inspect_err(
+					|e| debug!(sender=%user_id, ?e, "Failed to send a dummy event via user"),
+				)
+				.is_ok()
+			{
+				break;
+			}
+		}
+		let mut squash_timings = self.last_extremity_squash.write();
+		squash_timings.insert(room_id.to_owned(), Instant::now());
+	}
 }
diff --git a/src/service/rooms/event_handler/mod.rs b/src/service/rooms/event_handler/mod.rs
index 5025147f5..616dae3a4 100644
--- a/src/service/rooms/event_handler/mod.rs
+++ b/src/service/rooms/event_handler/mod.rs
@@ -25,6 +25,7 @@ use crate::{Dep, globals, rooms, sending, server_keys};
 pub struct Service {
 	pub mutex_federation: RoomMutexMap,
 	pub federation_handletime: SyncRwLock<HandleTimeMap>,
+	pub last_extremity_squash: SyncRwLock<HashMap<OwnedRoomId, Instant>>,
 	services: Services,
 	server_shutdown: Notify,
 }
@@ -55,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()),
 			services: Services {
 				globals: args.depend::<globals::Service>("globals"),
 				sending: args.depend::<sending::Service>("sending"),
@@ -90,6 +92,11 @@ impl crate::Service for Service {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 
 	fn interrupt(&self) { self.server_shutdown.notify_waiters(); }
+
+	async fn clear_cache(&self) {
+		let mut squashes = self.last_extremity_squash.write();
+		squashes.clear();
+	}
 }
 
 impl Service {