refactor: fix deadlocks in ingress system

packages/core_new/src/api/ingress.rs

@@ -1,5 +1,6 @@
 use crate::{store::{ingress::{IngressId, IngressType}, State}, LauncherResult};
 
+#[tracing::instrument]
 pub async fn init_ingress(
 	ingress_type: IngressType,
 	message: &str,
@@ -13,25 +14,43 @@ pub async fn init_ingress(
 	Ok(id)
 }
 
-pub async fn send_ingress(id: IngressId) -> LauncherResult<()> {
+#[tracing::instrument]
+pub async fn send_ingress(id: &IngressId, increment: f64) -> LauncherResult<()> {
 	let state = State::get().await?;
 	let processor = &state.ingress_processor;
 
-	processor.send(id.0).await?;
+	processor.send(id.0, increment).await?;
 	Ok(())
 }
 
-pub async fn update_ingress(id: IngressId, current: f64) -> LauncherResult<()> {
-	let state = State::get().await?;
-	let processor = &state.ingress_processor;
+#[cfg(test)]
+pub mod tests {
+    use std::time::Duration;
 
-	processor.update(id.0, current).await
-}
+    use crate::{api::{ingress::{init_ingress, send_ingress}, proxy::proxy_cli::ProxyCli}, initialize_core, store::ingress::IngressType, LauncherResult};
 
-pub async fn finish_ingress(id: IngressId) -> LauncherResult<()> {
-	let state = State::get().await?;
-	let processor = &state.ingress_processor;
 
-	processor.finish(id.0).await?;
-	Ok(())
+	#[tokio::test]
+	pub async fn create_and_update_ingress() -> LauncherResult<()> {
+		assert!(cfg!(feature = "cli"));
+
+		initialize_core(ProxyCli::new()).await?;
+
+		let id = init_ingress(IngressType::Download { file_name: "Some-Mod-1.8.9.jar".into() }, "This is a test message", 100.0).await?;
+		tokio::time::sleep(Duration::from_millis(2350)).await;
+
+		send_ingress(&id, 10.0).await?;
+		tokio::time::sleep(Duration::from_millis(2500)).await;
+
+		send_ingress(&id, 20.0).await?;
+		tokio::time::sleep(Duration::from_millis(2500)).await;
+
+		send_ingress(&id, 70.0).await?;
+		tokio::time::sleep(Duration::from_millis(2500)).await;
+
+		drop(id);
+
+		Ok(())
+	}
+
 }

packages/core_new/src/api/proxy/proxy_cli.rs

@@ -4,7 +4,7 @@ use indicatif::ProgressBar;
 use tokio::sync::RwLock;
 use uuid::Uuid;
 
-use crate::{store::{ingress::IngressPayload, State}, LauncherResult};
+use crate::{constants::CLI_TOTAL_INGRESS, store::ingress::IngressPayload, LauncherResult};
 
 use super::LauncherProxy;
 
@@ -24,21 +24,33 @@ impl ProxyCli {
 impl LauncherProxy for ProxyCli {
 
 	async fn send_ingress(&self, ingress: IngressPayload) -> LauncherResult<()> {
-		let feeds = self.ingress_feeds.write().await;
+		let mut feeds = self.ingress_feeds.write().await;
+
+		let completed = ingress.percent.is_none();
 
 		if let Some(progress) = feeds.get(&ingress.id) {
-			// progress.set_position(ingress.percent as u64);
-			// TODO: customize progress bar
-		} else {
-			let progress = ProgressBar::new(ingress.total as u64);
+			if completed {
+				progress.finish();
+			} else {
+				#[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
+				let new_pos = (ingress.percent.unwrap() * CLI_TOTAL_INGRESS as f64).round() as u64;
+				progress.set_position(new_pos);
+			}
+		} else if !completed {
+			let progress = ProgressBar::new(CLI_TOTAL_INGRESS);
 			progress.set_message(ingress.message.clone());
 			progress.set_position(0);
+			progress.set_style(indicatif::ProgressStyle::default_bar().template(
+                "{spinner:.green}, [{elapsed_precise}] [{bar:.lime/green}] {pos}/{len} {msg}"
+            ).unwrap().progress_chars("#>-"));
+
+			feeds.insert(ingress.id, progress);
 		}
 		Ok(())
 	}
 
 	async fn send_message(&self, message: crate::api::proxy::message::MessagePayload) -> LauncherResult<()> {
-		println!("[{:?}] {:?}", message.level, message.message);
+		println!("[{:?}] {}", message.level, message.message);
 		Ok(())
 	}
 

packages/core_new/src/lib.rs

@@ -6,5 +6,17 @@ pub mod api;
 mod error;
 mod logger;
 
+use api::proxy::LauncherProxy;
 pub use error::{LauncherResult, LauncherError};
-pub use logger::start_logger;
+pub use logger::start_logger;
+use store::{proxy::ProxyState, Dirs, State};
+
+pub async fn initialize_core(proxy_backend: impl LauncherProxy + 'static) -> LauncherResult<()> {
+	Dirs::get().await?;
+	start_logger().await;
+
+	ProxyState::initialize(proxy_backend).await?;
+	State::get().await?;
+
+	Ok(())
+}

packages/core_new/src/logger.rs

@@ -2,12 +2,12 @@
 // This will log to the console, and will not log to a file
 #[cfg(debug_assertions)]
 pub async fn start_logger() {
-    use tracing_subscriber::prelude::*;
+    use tracing_subscriber::{prelude::*, util::SubscriberInitExt};
 
     let filter = tracing_subscriber::EnvFilter::try_from_default_env()
-		.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(format!("{}=debug", env!("CARGO_PKG_NAME"))));
+		.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(format!("{}=info", env!("CARGO_PKG_NAME"))));
 
-    let subscriber = tracing_subscriber::registry()
+    tracing_subscriber::registry()
 	    .with(
 			tracing_subscriber::fmt::layer()
 				.with_ansi(true)
@@ -18,10 +18,8 @@ pub async fn start_logger() {
 				.pretty()
 		)
         .with(filter)
-        .with(tracing_error::ErrorLayer::default());
-
-	tracing::subscriber::set_global_default(subscriber)
-        .expect("setting default subscriber failed");
+        .with(tracing_error::ErrorLayer::default())
+		.init();
 }
 
 // Handling for the live production logging
@@ -60,7 +58,7 @@ pub async fn start_logger() {
     let filter = tracing_subscriber::EnvFilter::try_from_default_env()
         .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(format!("{}=info", env!("CARGO_PKG_NAME"))));
 
-    let subscriber = tracing_subscriber::registry()
+    tracing_subscriber::registry()
         .with(
             tracing_subscriber::fmt::layer()
 				.compact()
@@ -69,8 +67,6 @@ pub async fn start_logger() {
                 .with_timer(ChronoLocal::rfc_3339()),
         )
         .with(filter)
-        .with(tracing_error::ErrorLayer::default());
-
-    tracing::subscriber::set_global_default(subscriber)
-        .expect("Setting default subscriber failed");
+        .with(tracing_error::ErrorLayer::default())
+		.init();
 }

packages/core_new/src/store/db.rs

@@ -26,6 +26,7 @@ pub async fn create_pool_from_path(path: PathBuf) -> LauncherResult<SqlitePool>
 	Ok(pool)
 }
 
+#[tracing::instrument]
 pub async fn create_pool() -> LauncherResult<SqlitePool> {
 	let path = Dirs::get().await?.db_file();
 	create_pool_from_path(path).await

packages/core_new/src/store/ingress.rs

@@ -21,10 +21,15 @@ pub enum IngressError {
 
 impl IngressProcessor {
 	#[must_use]
+	#[tracing::instrument]
 	pub fn new() -> Self {
-		Self {
+		let processor = Self {
 			ingress_feeds: RwLock::new(HashMap::new()),
-		}
+		};
+
+		tracing::debug!("created ingress processor");
+
+		processor
 	}
 
 	pub async fn create(&self, ingress_type: IngressType, message: String, total: f64) -> LauncherResult<IngressId> {
@@ -42,24 +47,21 @@ impl IngressProcessor {
 
 		feeds.insert(uuid, ingress);
 
-		self.send(uuid).await?;
+		// Drop the write lock to prevent deadlock
+		drop(feeds);
+
+		self.send(uuid, 0.0).await?;
 
 		Ok(IngressId(uuid))
 	}
 
-	pub async fn update(&self, id: Uuid, current: f64) -> LauncherResult<()> {
+	pub async fn send(&self, id: Uuid, increment: f64) -> LauncherResult<()> {
 		let mut feeds = self.ingress_feeds.write().await;
-		let ingress = feeds.get_mut(&id).ok_or_else(|| IngressError::NotFound)?;
-		ingress.current = current;
-
-		Ok(())
-	}
-
-	pub async fn send(&self, id: Uuid) -> LauncherResult<()> {
-		let feeds = self.ingress_feeds.read().await;
-		let ingress = feeds.get(&id).ok_or_else(|| IngressError::NotFound)?;
+		let ingress = feeds.get_mut(&id).ok_or(IngressError::NotFound)?;
 
 		let proxy = ProxyState::get()?;
+
+		ingress.current += increment;
 		proxy.send_ingress(IngressPayload {
 			id: ingress.id.0,
 			message: ingress.message.clone(),
@@ -71,9 +73,9 @@ impl IngressProcessor {
 		Ok(())
 	}
 
-	pub async fn finish(&self, uuid: Uuid) -> LauncherResult<Ingress> {
+	pub async fn remove(&self, uuid: Uuid) -> LauncherResult<Ingress> {
 		let mut feeds = self.ingress_feeds.write().await;
-		Ok(feeds.remove(&uuid).ok_or_else(|| IngressError::NotFound)?)
+		Ok(feeds.remove(&uuid).ok_or(IngressError::NotFound)?)
 	}
 }
 
@@ -119,7 +121,7 @@ impl Drop for IngressId {
 				let state = State::get().await?;
 				let processor = &state.ingress_processor;
 
-				let ingress = processor.finish(ingress_uuid).await?;
+				let ingress = processor.remove(ingress_uuid).await?;
 
 				let proxy = ProxyState::get()?;
 

packages/core_new/src/store/proxy.rs

@@ -20,11 +20,11 @@ impl Deref for ProxyState {
 }
 
 impl ProxyState {
-	pub async fn initialize(proxy: Box<dyn LauncherProxy>) -> LauncherResult<Arc<Self>> {
+	pub async fn initialize(proxy: impl LauncherProxy + 'static) -> LauncherResult<Arc<Self>> {
 		PROXY_STATE
 			.get_or_try_init(|| async {
 				Ok(Arc::new(Self {
-					inner: proxy,
+					inner: Box::new(proxy),
 				}))
 			})
 			.await

packages/core_new/src/store/settings.rs

@@ -32,6 +32,7 @@ impl Default for Settings {
 }
 
 impl Settings {
+	#[tracing::instrument]
 	pub async fn new() -> Self {
 		match Self::read().await {
 			Ok(settings) => settings,

packages/core_new/src/store/state.rs

@@ -41,6 +41,7 @@ impl State {
 		LAUNCHER_STATE.initialized()
 	}
 
+	#[tracing::instrument]
 	async fn initialize() -> LauncherResult<RwLock<Self>> {
 		let ingress_processor = IngressProcessor::new();
 		let settings = Settings::new().await;