add conf item for exact amount of startup netburst.

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

conduwuit-example.toml

@@ -308,6 +308,18 @@ url_preview_check_root_domain = false
 # No default.
 #auto_join_rooms = []
 
+# Retry failed and incomplete messages to remote servers immediately upon startup. This is called bursting.
+# If this is disabled, said messages may not be delivered until more messages are queued for that server.
+# Do not change this option unless server resources are extremely limited or the scale of the server's
+# deployment is huge. Do not disable this unless you know what you are doing.
+#startup_netburst = true
+
+# Limit the startup netburst to the most recent (default: 50) messages queued for each remote server. All older
+# messages are dropped and not reattempted. The `startup_netburst` option must be enabled for this value to have
+# any effect. Do not change this value unless you know what you are doing. Set this value to -1 to reattempt
+# every message without trimming the queues; this may consume significant disk. Set this value to 0 to drop all
+# messages without any attempt at redelivery.
+#startup_netburst_keep = 50
 
 
 ### Generic database options

src/config/mod.rs

@@ -245,6 +245,8 @@ pub struct Config {
 
 	#[serde(default = "true_fn")]
 	pub startup_netburst: bool,
+	#[serde(default = "default_startup_netburst_keep")]
+	pub startup_netburst_keep: i64,
 
 	#[serde(default)]
 	pub block_non_admin_invites: bool,
@@ -733,3 +735,5 @@ fn default_url_preview_max_spider_size() -> usize {
 fn default_new_user_displayname_suffix() -> String { "🏳️‍⚧️".to_owned() }
 
 fn default_sentry_traces_sample_rate() -> f32 { 0.15 }
+
+fn default_startup_netburst_keep() -> i64 { 50 }

src/service/sending/mod.rs

@@ -45,6 +45,7 @@ pub struct Service {
 	pub sender: mpsc::UnboundedSender<(OutgoingKind, SendingEventType, Vec<u8>)>,
 	receiver: Mutex<mpsc::UnboundedReceiver<(OutgoingKind, SendingEventType, Vec<u8>)>>,
 	startup_netburst: bool,
+	startup_netburst_keep: i64,
 	timeout: u64,
 }
 
@@ -78,6 +79,7 @@ impl Service {
 			receiver: Mutex::new(receiver),
 			maximum_requests: Arc::new(Semaphore::new(config.max_concurrent_requests as usize)),
 			startup_netburst: config.startup_netburst,
+			startup_netburst_keep: config.startup_netburst_keep,
 			timeout: config.sender_timeout,
 		})
 	}
@@ -284,8 +286,10 @@ impl Service {
 					.entry(outgoing_kind.clone())
 					.or_default();
 
-				if entry.len() > 30 {
+				if self.startup_netburst_keep >= 0
-					warn!("Dropping some current events: {:?} {:?} {:?}", key, outgoing_kind, event);
+					&& entry.len() >= usize::try_from(self.startup_netburst_keep).unwrap()
+				{
+					warn!("Dropping unsent event {:?} {:?}", outgoing_kind, String::from_utf8_lossy(&key),);
 					self.db.delete_active_request(key)?;
 					continue;
 				}