feat: Add jitter to extremity squasher

src/core/utils/time.rs

@@ -1,7 +1,10 @@
 pub mod exponential_backoff;
+pub mod jitter;
 
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+pub use jitter::jitter;
+
 use crate::{Result, err};
 
 #[inline]

src/core/utils/time/jitter.rs

@@ -0,0 +1,16 @@
+use std::{ops::RangeInclusive, time::Duration};
+
+/// Returns a `Duration` that is jittered by a random percentage of the base
+/// duration. The jitter is applied as a random percentage in the range of
+/// `-jitter_range` to `jitter_range`.
+///
+/// # Example
+/// ```
+/// use conduwuit_core::utils::time::jitter;
+/// let sleep_duration = jitter(Duration::from_secs(1), -10..=10);
+/// // Adds a jitter of between -10% and 10% to the duration.
+/// ```
+#[must_use]
+pub fn jitter(base: Duration, jitter_range: RangeInclusive<f64>) -> Duration {
+	base.mul_f64(1.0 + (rand::random_range(jitter_range)) / 100.0)
+}

src/service/announcements/mod.rs

@@ -18,7 +18,11 @@
 use std::{sync::Arc, time::Duration};
 
 use async_trait::async_trait;
-use conduwuit::{Result, Server, debug, error, utils::response::LimitReadExt, warn};
+use conduwuit::{
+	Result, Server, debug, error,
+	utils::{response::LimitReadExt, time::jitter},
+	warn,
+};
 use database::{Deserialized, Map};
 use ruma::events::{Mentions, room::message::RoomMessageEventContent};
 use serde::Deserialize;
@@ -98,10 +102,7 @@ impl crate::Service for Service {
 				.ok();
 		}
 
-		let first_check_jitter = {
-			let jitter_percent = rand::random_range(-50.0..=10.0);
-			self.interval.mul_f64(1.0 + jitter_percent / 100.0)
-		};
+		let first_check_jitter = jitter(self.interval, -50.0..=10.0);
 
 		let mut i = interval(self.interval);
 		i.set_missed_tick_behavior(MissedTickBehavior::Delay);

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -5,7 +5,7 @@ use std::{
 
 use conduwuit::{
 	Err, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, defer, err, error,
-	info, matrix::PartialPdu, result::DebugInspect, trace, warn,
+	info, matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn,
 };
 use futures::{
 	FutureExt, StreamExt,
@@ -338,7 +338,7 @@ impl super::Service {
 
 			let mut closing = false;
 
-			let waker = tokio::time::sleep(Duration::from_mins(2));
+			let waker = tokio::time::sleep(jitter(Duration::from_mins(2), -25.0..=25.0));
 			tokio::pin!(waker);
 
 			loop {
@@ -348,7 +348,7 @@ impl super::Service {
 							latest_extremity_count = Some(extremities_count);
 							non_dummy_event = non_dummy_event || !is_dummy_event;
 							#[allow(clippy::arithmetic_side_effects)]
-							waker.as_mut().reset(tokio::time::Instant::now() + Duration::from_mins(1));
+							waker.as_mut().reset(tokio::time::Instant::now() + jitter(Duration::from_mins(1), -50.0..=50.0));
 						} else {
 							{let mut map = service.extremity_squashers.write();
 							if let Some(tx) = map.get(&room_id) && tx.is_closed() {