feat: Automatically squash extremities when they exceed a threshold

Attempts to tackle #1844

conduwuit-example.toml

@@ -645,6 +645,14 @@
 #
 #default_room_acl_deny =
 
+# The number of forward extremities to tolerate in a room before
+# attempting to manually squash them with a "dummy event". Setting this
+# above 20 will hinder its efficacy, and setting it below 5 will cause
+# more dummy events to be sent than necessary (which increases federation
+# traffic).
+#
+#dummy_event_threshold = 10
+
 # Enable OpenTelemetry OTLP tracing export. This replaces the deprecated
 # Jaeger exporter. Traces will be sent via OTLP to a collector (such as
 # Jaeger) that supports the OpenTelemetry Protocol.

src/core/config/mod.rs

@@ -781,6 +781,16 @@ pub struct Config {
 	/// a substitute for moderation bots.
 	pub default_room_acl_deny: Option<Vec<String>>,
 
+	/// The number of forward extremities to tolerate in a room before
+	/// attempting to manually squash them with a "dummy event". Setting this
+	/// above 20 will hinder its efficacy, and setting it below 5 will cause
+	/// more dummy events to be sent than necessary (which increases federation
+	/// traffic).
+	///
+	/// default: 10
+	#[serde(default = "default_extremity_threshold")]
+	pub dummy_event_threshold: u8,
+
 	/// display: nested
 	#[serde(default)]
 	pub well_known: WellKnownConfig,
@@ -2652,6 +2662,8 @@ fn default_rocksdb_stats_level() -> u8 { 1 }
 #[inline]
 pub fn default_default_room_version() -> RoomVersionId { RoomVersionId::V12 }
 
+fn default_extremity_threshold() -> u8 { 10 }
+
 fn default_ip_range_denylist() -> Vec<String> {
 	vec![
 		"127.0.0.0/8".to_owned(),

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -1,11 +1,11 @@
 use std::{collections::BTreeMap, time::Instant};
 
 use conduwuit::{
-	Err, Event, PduEvent, Result, debug, debug_error, debug_info, defer, err, error, info,
-	result::DebugInspect, trace, warn,
+	Err, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, defer, err, error,
+	info, matrix::PartialPdu, result::DebugInspect, trace, warn,
 };
 use futures::{
-	FutureExt,
+	FutureExt, StreamExt,
 	future::{OptionFuture, try_join4},
 };
 use ruma::{
@@ -15,6 +15,7 @@
 		room::member::{MembershipState, RoomMemberEventContent},
 	},
 };
+use serde_json::{json, value::to_raw_value};
 
 use crate::rooms::timeline::{RawPduId, pdu_fits};
 
@@ -259,7 +260,60 @@ pub async fn handle_incoming_pdu<'a>(
 			})?;
 
 		// Done with prev events, now handling the incoming event
-		self.upgrade_outlier_to_timeline_pdu(incoming_pdu, val, create_event, origin, room_id)
-			.await
+		let pdu_id = self
+			.upgrade_outlier_to_timeline_pdu(incoming_pdu, val, create_event, origin, room_id)
+			.await?;
+
+		let extremities_count = self
+			.services
+			.state
+			.get_forward_extremities(room_id)
+			.count()
+			.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
+			while let Some(user_id) = self
+				.services
+				.state_cache
+				.local_users_in_room(room_id)
+				.next()
+				.await
+			{
+				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(),
+							content: to_raw_value(&json!({})).expect("john json"),
+							unsigned: None,
+							state_key: None,
+							redacts: None,
+							timestamp: None,
+						},
+						&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;
+				}
+			}
+		}
+
+		Ok(pdu_id)
 	}
 }