diff --git a/packages/app-lib/src/api/friends.rs b/packages/app-lib/src/api/friends.rs index 0fe883e01d..cfc70f6e3c 100644 --- a/packages/app-lib/src/api/friends.rs +++ b/packages/app-lib/src/api/friends.rs @@ -12,7 +12,7 @@ pub async fn friends() -> crate::Result> { pub async fn friend_statuses() -> crate::Result> { let state = crate::State::get().await?; - let statuses = state.friends_socket.friend_statuses(); + let statuses = state.presence.friends_socket.friend_statuses(); Ok(statuses) } diff --git a/packages/app-lib/src/api/instance/files.rs b/packages/app-lib/src/api/instance/files.rs index 2f0830c7f5..4a186974da 100644 --- a/packages/app-lib/src/api/instance/files.rs +++ b/packages/app-lib/src/api/instance/files.rs @@ -192,7 +192,7 @@ pub async fn write_instance_file( create_only: bool, ) -> crate::Result<()> { let state = State::get().await?; - let _instance = state.lock_instance_content(instance_id).await; + let _instance = state.instance_locks.lock_content(instance_id).await; let _files = state.content_store.files_lock.lock().await; let destination = resolve(&state, instance_id, path, true) .await? @@ -238,7 +238,7 @@ pub async fn create_instance_directory( path: &str, ) -> crate::Result<()> { let state = State::get().await?; - let _instance = state.lock_instance_content(instance_id).await; + let _instance = state.instance_locks.lock_content(instance_id).await; let _files = state.content_store.files_lock.lock().await; let destination = resolve(&state, instance_id, path, true).await?; fs::create_dir(destination.path()).await?; @@ -251,7 +251,7 @@ pub async fn rename_instance_file( destination: &str, ) -> crate::Result<()> { let state = State::get().await?; - let _instance = state.lock_instance_content(instance_id).await; + let _instance = state.instance_locks.lock_content(instance_id).await; let _files = state.content_store.files_lock.lock().await; let source = resolve(&state, instance_id, source, true).await?; let destination = resolve(&state, instance_id, destination, true).await?; @@ -270,7 +270,7 @@ pub async fn delete_instance_file( recursive: bool, ) -> crate::Result<()> { let state = State::get().await?; - let _instance = state.lock_instance_content(instance_id).await; + let _instance = state.instance_locks.lock_content(instance_id).await; let _files = state.content_store.files_lock.lock().await; let path = resolve(&state, instance_id, path, true).await?; if fs::symlink_metadata(path.path()).await?.is_dir() { diff --git a/packages/app-lib/src/api/instance/screenshots/operations.rs b/packages/app-lib/src/api/instance/screenshots/operations.rs index 5d7b618fb8..6d3e8f6d5d 100644 --- a/packages/app-lib/src/api/instance/screenshots/operations.rs +++ b/packages/app-lib/src/api/instance/screenshots/operations.rs @@ -18,6 +18,7 @@ use crate::state::instances::adapters::sqlite::{ instance_rows::{self, InstanceScreenshotSource}, screenshot_rows, }; +use crate::state::{DirectoryInfo, InstanceLocks}; use crate::util::fetch::sha1_file_async; use crate::util::io::{self, IOError}; use crate::{ErrorKind, State}; @@ -103,12 +104,12 @@ async fn list_source_screenshot_sets( } async fn lock_instance_screenshots<'a>( - state: &State, + instance_locks: &InstanceLocks, instance_ids: impl IntoIterator, ) -> Vec> { let mut locks = Vec::new(); for instance_id in instance_ids { - locks.push(state.lock_instance_screenshots(instance_id).await); + locks.push(instance_locks.lock_screenshots(instance_id).await); } locks } @@ -122,7 +123,8 @@ pub async fn delete_screenshots(keys: &[ScreenshotKey]) -> crate::Result<()> { .collect::>(); instance_ids.sort_unstable(); instance_ids.dedup(); - let _locks = lock_instance_screenshots(&state, instance_ids).await; + let _locks = + lock_instance_screenshots(&state.instance_locks, instance_ids).await; for key in keys { io::remove_file(get_screenshot_path(key).await?).await?; @@ -224,14 +226,16 @@ pub async fn move_screenshots( .collect::>(); instance_ids.sort_unstable(); instance_ids.dedup(); - let _locks = lock_instance_screenshots(&state, instance_ids).await; + let _locks = + lock_instance_screenshots(&state.instance_locks, instance_ids).await; let target_source = instance_rows::get_instance_screenshot_source( target_instance_id, &state.pool, ) .await? .ok_or(ErrorKind::InputError("unknown target instance".to_string()))?; - let target_dir = source_screenshots_dir(&state, &target_source).await?; + let target_dir = + source_screenshots_dir(&state.directories, &target_source).await?; io::create_dir_all(&target_dir).await?; ensure_directory_is_not_symlink(&target_dir).await?; @@ -285,7 +289,8 @@ pub async fn get_screenshot_path( ) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; - let screenshots_dir = source_screenshots_dir(&state, &source).await?; + let screenshots_dir = + source_screenshots_dir(&state.directories, &source).await?; let canonical_dir = tokio::fs::canonicalize(&screenshots_dir) .await .map_err(|error| IOError::with_path(error, &screenshots_dir))?; @@ -327,9 +332,9 @@ pub async fn save_edited_screenshot( ) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; - let _lock = state.lock_instance_screenshots(&source.id).await; + let _lock = state.instance_locks.lock_screenshots(&source.id).await; - let scanned = scan_source_screenshots(&state, &source).await?; + let scanned = scan_source_screenshots(&state.directories, &source).await?; let current = reconcile_source_screenshots(&state, &source, scanned).await?; let source_screenshot = current @@ -374,7 +379,8 @@ pub async fn save_edited_screenshot( ))); } - let screenshots_dir = source_screenshots_dir(&state, &source).await?; + let screenshots_dir = + source_screenshots_dir(&state.directories, &source).await?; let (target_path, copy_group) = match mode { ScreenshotEditSaveMode::CreateCopy => ( available_target_path(&screenshots_dir, &source_row.file_name) @@ -407,7 +413,7 @@ pub async fn save_edited_screenshot( screenshot_rows::update_screenshot(&source_row, &mut tx).await?; tx.commit().await?; } - let scanned = scan_source_screenshots(&state, &source).await?; + let scanned = scan_source_screenshots(&state.directories, &source).await?; let reconciled = reconcile_source_screenshots(&state, &source, scanned).await?; let mut saved = reconciled @@ -478,10 +484,10 @@ pub async fn save_edited_screenshot( } pub(super) async fn source_screenshots_dir( - state: &State, + directories: &DirectoryInfo, source: &InstanceScreenshotSource, ) -> crate::Result { - let instance_dir = state.directories.instances_dir().join(&source.path); + let instance_dir = directories.instances_dir().join(&source.path); let canonical_instance_dir = tokio::fs::canonicalize(&instance_dir) .await diff --git a/packages/app-lib/src/api/instance/screenshots/reconciliation.rs b/packages/app-lib/src/api/instance/screenshots/reconciliation.rs index e5c8537f43..b9bc5cd720 100644 --- a/packages/app-lib/src/api/instance/screenshots/reconciliation.rs +++ b/packages/app-lib/src/api/instance/screenshots/reconciliation.rs @@ -2,6 +2,7 @@ use super::operations::{ InstanceScreenshot, has_png_extension, sort_screenshots, source_screenshots_dir, }; +use crate::state::DirectoryInfo; use crate::state::instances::adapters::sqlite::{ instance_rows::{self, InstanceScreenshotSource}, screenshot_rows::{self, ScreenshotRow}, @@ -80,8 +81,8 @@ pub(super) async fn list_source_screenshots( state: &State, source: InstanceScreenshotSource, ) -> crate::Result> { - let _lock = state.lock_instance_screenshots(&source.id).await; - let scanned = scan_source_screenshots(state, &source).await?; + let _lock = state.instance_locks.lock_screenshots(&source.id).await; + let scanned = scan_source_screenshots(&state.directories, &source).await?; reconcile_source_screenshots(state, &source, scanned).await } @@ -101,10 +102,10 @@ pub(crate) async fn reconcile_screenshots( } pub(super) async fn scan_source_screenshots( - state: &State, + directories: &DirectoryInfo, source: &InstanceScreenshotSource, ) -> crate::Result> { - let instance_dir = state.directories.instances_dir().join(&source.path); + let instance_dir = directories.instances_dir().join(&source.path); if !tokio::fs::try_exists(&instance_dir) .await .map_err(|error| IOError::with_path(error, &instance_dir))? @@ -112,7 +113,7 @@ pub(super) async fn scan_source_screenshots( return Ok(Vec::new()); } - let screenshots_dir = source_screenshots_dir(state, source).await?; + let screenshots_dir = source_screenshots_dir(directories, source).await?; tokio::task::spawn_blocking(move || scan_screenshots_dir(&screenshots_dir)) .await? } diff --git a/packages/app-lib/src/api/instance/shared/invites.rs b/packages/app-lib/src/api/instance/shared/invites.rs index b7bd7260ca..a984d5fb61 100644 --- a/packages/app-lib/src/api/instance/shared/invites.rs +++ b/packages/app-lib/src/api/instance/shared/invites.rs @@ -25,7 +25,8 @@ pub async fn invite_shared_instance_users( user_ids: Vec, ) -> crate::Result { let state = State::get().await?; - let _shared_instance_lock = state.lock_shared_instance(instance_id).await; + let _shared_instance_lock = + state.instance_locks.lock_shared(instance_id).await; let (metadata, attachment) = shared_instance_for_invites(instance_id, user_ids.len(), &state) .await?; @@ -65,7 +66,8 @@ pub async fn create_shared_instance_invite_link( replace_invite_id: Option, ) -> crate::Result { let state = State::get().await?; - let _shared_instance_lock = state.lock_shared_instance(instance_id).await; + let _shared_instance_lock = + state.instance_locks.lock_shared(instance_id).await; let (metadata, attachment) = shared_instance_for_invites(instance_id, 0, &state).await?; ensure_owner(&attachment)?; @@ -150,7 +152,8 @@ pub async fn revoke_shared_instance_invite( invite_id: String, ) -> crate::Result<()> { let state = State::get().await?; - let _shared_instance_lock = state.lock_shared_instance(instance_id).await; + let _shared_instance_lock = + state.instance_locks.lock_shared(instance_id).await; let Some(attachment) = shared_attachment(instance_id, &state).await? else { return Ok(()); }; @@ -167,7 +170,8 @@ pub async fn remove_shared_instance_users( has_pending_recipients: bool, ) -> crate::Result { let state = State::get().await?; - let _shared_instance_lock = state.lock_shared_instance(instance_id).await; + let _shared_instance_lock = + state.instance_locks.lock_shared(instance_id).await; let Some(attachment) = shared_attachment(instance_id, &state).await? else { return Ok(SharedInstanceUsers::empty()); }; diff --git a/packages/app-lib/src/api/instance/synced_options/command_history.rs b/packages/app-lib/src/api/instance/synced_options/command_history.rs index 59d19237a3..5ee1a4fce5 100644 --- a/packages/app-lib/src/api/instance/synced_options/command_history.rs +++ b/packages/app-lib/src/api/instance/synced_options/command_history.rs @@ -15,7 +15,7 @@ const COMMAND_HISTORY_LIMIT: usize = 50; pub async fn get_command_history() -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let path = command_history_path(&state); if !path.exists() { return Ok(String::new()); @@ -25,7 +25,7 @@ pub async fn get_command_history() -> crate::Result { pub async fn set_command_history(contents: &str) -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; create_synced_directories(&state).await?; let normalized = normalize_command_history(contents); io::write(command_history_path(&state), normalized.as_bytes()).await?; diff --git a/packages/app-lib/src/api/instance/synced_options/game_options/launch_overrides.rs b/packages/app-lib/src/api/instance/synced_options/game_options/launch_overrides.rs index dc0dd3dc17..c0ea35f311 100644 --- a/packages/app-lib/src/api/instance/synced_options/game_options/launch_overrides.rs +++ b/packages/app-lib/src/api/instance/synced_options/game_options/launch_overrides.rs @@ -26,7 +26,7 @@ pub(super) fn currently_launcher_owned_keys( /// Writes canonical settings to `options.txt` before launch. pub async fn sync_before_launch(instance_id: &str) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; @@ -48,7 +48,7 @@ pub async fn apply_launcher_overrides( return Ok(()); } let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; diff --git a/packages/app-lib/src/api/instance/synced_options/game_options/local_settings_editor.rs b/packages/app-lib/src/api/instance/synced_options/game_options/local_settings_editor.rs index 59c2485ebc..b5efbda201 100644 --- a/packages/app-lib/src/api/instance/synced_options/game_options/local_settings_editor.rs +++ b/packages/app-lib/src/api/instance/synced_options/game_options/local_settings_editor.rs @@ -266,7 +266,7 @@ pub async fn get_config( instance_id: &str, ) -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; load_editor(instance_id, &state).await } @@ -275,7 +275,7 @@ pub async fn preview_changes( request: UpdateGameSettingsRequest, ) -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let (metadata, document, input_bytes, controlled_keys) = load_instance_document(instance_id, &state).await?; let mut editor = build_editor_state( @@ -335,7 +335,7 @@ pub async fn save_changes( request: UpdateGameSettingsRequest, ) -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let (metadata, mut document, input_bytes, controlled_keys) = load_instance_document(instance_id, &state).await?; let current_editor = build_editor_state( diff --git a/packages/app-lib/src/api/instance/synced_options/game_options/locales/mod.rs b/packages/app-lib/src/api/instance/synced_options/game_options/locales/mod.rs index 0addbb155d..d2c1a9a25e 100644 --- a/packages/app-lib/src/api/instance/synced_options/game_options/locales/mod.rs +++ b/packages/app-lib/src/api/instance/synced_options/game_options/locales/mod.rs @@ -80,7 +80,7 @@ struct Candidate { pub(crate) fn queue_game_locale_index() { if let Some(state) = State::get_if_initialized() { - state.game_locale_indexer.notify.notify_one(); + state.synced_options.locales.notify.notify_one(); } else { tracing::warn!( "Game setting locales: cannot queue indexing before state initialization" @@ -90,16 +90,17 @@ pub(crate) fn queue_game_locale_index() { pub(crate) fn start_game_locale_indexer(state: Arc) { if state - .game_locale_indexer + .synced_options + .locales .started .swap(true, Ordering::AcqRel) { return; } - state.game_locale_indexer.notify.notify_one(); + state.synced_options.locales.notify.notify_one(); tokio::spawn(async move { loop { - state.game_locale_indexer.notify.notified().await; + state.synced_options.locales.notify.notified().await; tokio::time::sleep(std::time::Duration::from_millis(300)).await; if let Err(error) = index_installed_sources(&state).await { tracing::warn!(%error, "Game setting locales: indexing failed"); diff --git a/packages/app-lib/src/api/instance/synced_options/game_options/pack_updates.rs b/packages/app-lib/src/api/instance/synced_options/game_options/pack_updates.rs index e797bff340..ff95b169d4 100644 --- a/packages/app-lib/src/api/instance/synced_options/game_options/pack_updates.rs +++ b/packages/app-lib/src/api/instance/synced_options/game_options/pack_updates.rs @@ -293,7 +293,7 @@ pub async fn capture_pack_base( source: Option, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(crate::ErrorKind::InputError("unknown instance".to_string()))?; diff --git a/packages/app-lib/src/api/instance/synced_options/game_options/settings_editor.rs b/packages/app-lib/src/api/instance/synced_options/game_options/settings_editor.rs index 98cb379887..78de1778b9 100644 --- a/packages/app-lib/src/api/instance/synced_options/game_options/settings_editor.rs +++ b/packages/app-lib/src/api/instance/synced_options/game_options/settings_editor.rs @@ -195,7 +195,7 @@ pub(super) async fn load_settings_editor( pub async fn get_config() -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; load_settings_editor(&state).await } @@ -204,7 +204,7 @@ pub async fn preview_changes( request: UpdateGameSettingsRequest, ) -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; if request.expected_catalog_revision != CATALOG_REVISION { return load_settings_editor(&state).await; } @@ -277,7 +277,7 @@ pub async fn save_changes( request: UpdateGameSettingsRequest, ) -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; if !game_options_sync_is_enabled(&state.pool).await? { bail!(ErrorKind::InputError( "turn on game-settings sync before editing shared values" diff --git a/packages/app-lib/src/api/instance/synced_options/game_options/source_selection.rs b/packages/app-lib/src/api/instance/synced_options/game_options/source_selection.rs index 60aef12bd9..ee562667fa 100644 --- a/packages/app-lib/src/api/instance/synced_options/game_options/source_selection.rs +++ b/packages/app-lib/src/api/instance/synced_options/game_options/source_selection.rs @@ -323,7 +323,7 @@ pub(in crate::api::instance) async fn initialize_from_source_instance( pub async fn list_sync_sources() -> crate::Result> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let instances = crate::state::list_instances(&state.pool).await?; let mut sources = Vec::with_capacity(instances.len()); for metadata in instances { diff --git a/packages/app-lib/src/api/instance/synced_options/orchestration.rs b/packages/app-lib/src/api/instance/synced_options/orchestration.rs index 57f5262f8c..09b32fb668 100644 --- a/packages/app-lib/src/api/instance/synced_options/orchestration.rs +++ b/packages/app-lib/src/api/instance/synced_options/orchestration.rs @@ -123,7 +123,7 @@ pub async fn get_global_options() -> crate::Result { pub async fn get_initialized_options() -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let mut options = GlobalSyncedOptions::default(); for option in SyncedOption::ALL { @@ -348,7 +348,7 @@ pub async fn set_global_option( )); } let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; pending::cancel(option, None, &state).await?; set_global_option_with_state(option, enabled, base_instance_id, &state) .await @@ -670,7 +670,7 @@ pub async fn set_instance_option( )); } let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; @@ -768,7 +768,7 @@ pub async fn get_instance_option_join_preview( option: SyncedOption, ) -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; @@ -919,7 +919,7 @@ async fn apply_pending_changes(state: &State) -> crate::Result<()> { pub async fn reconcile_all() -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; create_synced_directories(&state).await?; apply_pending_changes(&state).await?; let instances = crate::state::list_instances(&state.pool).await?; @@ -991,7 +991,7 @@ pub fn reconcile_instance( async fn reconcile_instance_inner(instance_id: &str) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; apply_pending_changes(&state).await?; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? @@ -1003,7 +1003,7 @@ pub(crate) async fn reconcile_instance_after_pack_update( instance_id: &str, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; @@ -1014,7 +1014,7 @@ pub(crate) async fn prepare_instance_update( instance_id: &str, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; @@ -1131,7 +1131,7 @@ pub async fn reconcile_changed_file( file_name: &str, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; apply_pending_changes(&state).await?; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? diff --git a/packages/app-lib/src/api/instance/synced_packs/operations.rs b/packages/app-lib/src/api/instance/synced_packs/operations.rs index 273485fd8f..2849c9d2ad 100644 --- a/packages/app-lib/src/api/instance/synced_packs/operations.rs +++ b/packages/app-lib/src/api/instance/synced_packs/operations.rs @@ -560,7 +560,7 @@ pub async fn upload_synced_pack( let size = bytes.len() as u64; let sha1 = crate::util::fetch::sha1_async(bytes.clone()).await?; let sha512 = cache_bytes(bytes, &state).await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; if !get_global_options().await?.get(option) { bail!(ErrorKind::InputError( "pack syncing was disabled while preparing the upload".to_owned(), @@ -613,7 +613,7 @@ pub async fn set_synced_pack_enabled( enabled: bool, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let mut library = read_library(&state).await?; let pack = library .packs @@ -626,7 +626,7 @@ pub async fn set_synced_pack_enabled( pub async fn remove_synced_pack(pack_id: &str) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let mut library = read_library(&state).await?; let pack = library .packs @@ -643,7 +643,7 @@ pub async fn desync_pack( mode: DesyncServerMode, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; diff --git a/packages/app-lib/src/api/instance/synced_packs/reconciliation.rs b/packages/app-lib/src/api/instance/synced_packs/reconciliation.rs index 3ea3549de3..44fd97361a 100644 --- a/packages/app-lib/src/api/instance/synced_packs/reconciliation.rs +++ b/packages/app-lib/src/api/instance/synced_packs/reconciliation.rs @@ -309,7 +309,7 @@ async fn prepare_pack( None, ) .await?; - downloaded.store_file(state).await? + downloaded.store_file(&state.content_store).await? } else { read_stored_file(pack, state).await? }; diff --git a/packages/app-lib/src/api/instance/synced_packs/storage.rs b/packages/app-lib/src/api/instance/synced_packs/storage.rs index bffa9b7672..e40f47d842 100644 --- a/packages/app-lib/src/api/instance/synced_packs/storage.rs +++ b/packages/app-lib/src/api/instance/synced_packs/storage.rs @@ -103,7 +103,7 @@ pub(super) async fn write_library( .await?; tokio::fs::rename(&temporary, &destination).await?; renamed = true; - state.pack_sync_worker.revision + state.synced_options.packs.revision .fetch_add(1, std::sync::atomic::Ordering::Release); } crate::state::content_store::sync_directory(&directory(state)).await?; @@ -236,7 +236,7 @@ pub(super) async fn read_stored_file( pub(crate) async fn migrate_store(state: &State) -> crate::Result<()> { let ids = { - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; read_library(state) .await? .packs @@ -247,7 +247,7 @@ pub(crate) async fn migrate_store(state: &State) -> crate::Result<()> { for id in ids { let _turn = state.content_store.legacy_migration_priority.read().await; let snapshot = { - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; read_library(state).await? }; let Some(expected) = snapshot.packs.get(&id) else { @@ -255,7 +255,7 @@ pub(crate) async fn migrate_store(state: &State) -> crate::Result<()> { }; let recovered = recover_legacy_pack(Some(&id), expected, &snapshot, state).await; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let mut library = read_library(state).await?; let Some(previous) = library.packs.get(&id).cloned() else { continue; @@ -295,7 +295,7 @@ pub(crate) async fn migrate_store(state: &State) -> crate::Result<()> { } } let cache_complete = cleanup_legacy_cache(state).await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; if tokio::fs::try_exists(directory(state)).await? { crate::state::content_store::sync_directory(&directory(state)).await?; } @@ -443,7 +443,7 @@ async fn cleanup_legacy_cache(state: &State) -> crate::Result { let mut entries = tokio::fs::read_dir(&legacy).await?; while let Some(entry) = entries.next_entry().await? { let _turn = state.content_store.legacy_migration_priority.read().await; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let library = read_library(state).await?; if !entry.file_type().await?.is_file() { continue; diff --git a/packages/app-lib/src/api/instance/synced_packs/worker.rs b/packages/app-lib/src/api/instance/synced_packs/worker.rs index fd97056728..4b1c605843 100644 --- a/packages/app-lib/src/api/instance/synced_packs/worker.rs +++ b/packages/app-lib/src/api/instance/synced_packs/worker.rs @@ -35,7 +35,8 @@ pub(crate) struct PackSyncWorker { pub(crate) fn queue_reconciliation(instance_id: &str) { if let Some(state) = State::get_if_initialized() { state - .pack_sync_worker + .synced_options + .packs .queue .lock() .instances @@ -46,7 +47,7 @@ pub(crate) fn queue_reconciliation(instance_id: &str) { pub(super) fn queue_all() { if let Some(state) = State::get_if_initialized() { - state.pack_sync_worker.queue.lock().all = true; + state.synced_options.packs.queue.lock().all = true; start(state); } } @@ -55,7 +56,7 @@ pub(crate) async fn flush(instance_id: &str) -> Result<(), Arc> { let state = State::get().await?; let (sender, receiver) = oneshot::channel(); { - let mut queue = state.pack_sync_worker.queue.lock(); + let mut queue = state.synced_options.packs.queue.lock(); if let Some(retry_at) = queue.retry_at.get(instance_id) { let remaining = retry_at.saturating_duration_since(Instant::now()); if !remaining.is_zero() { @@ -82,7 +83,7 @@ pub(crate) async fn flush(instance_id: &str) -> Result<(), Arc> { fn start(state: Arc) { { - let mut queue = state.pack_sync_worker.queue.lock(); + let mut queue = state.synced_options.packs.queue.lock(); if queue.running { return; } @@ -91,29 +92,36 @@ fn start(state: Arc) { tokio::spawn(async move { loop { tokio::time::sleep(std::time::Duration::from_millis(100)).await; - let all = - std::mem::take(&mut state.pack_sync_worker.queue.lock().all); + let all = std::mem::take( + &mut state.synced_options.packs.queue.lock().all, + ); if all { match crate::state::list_instances(&state.pool).await { Ok(instances) => { - state.pack_sync_worker.queue.lock().instances.extend( - instances - .into_iter() - .map(|metadata| metadata.instance.id), - ); + state + .synced_options + .packs + .queue + .lock() + .instances + .extend( + instances + .into_iter() + .map(|metadata| metadata.instance.id), + ); } Err(error) => { tracing::warn!( "Could not enumerate pack sync targets: {error}" ); - state.pack_sync_worker.queue.lock().all = true; + state.synced_options.packs.queue.lock().all = true; tokio::time::sleep(std::time::Duration::from_secs(1)) .await; } } } let (instance_id, mut pending) = { - let mut queue = state.pack_sync_worker.queue.lock(); + let mut queue = state.synced_options.packs.queue.lock(); let next = queue .waiters .keys() @@ -139,7 +147,7 @@ fn start(state: Arc) { if let Err(error) = &result { match error.downcast_ref::() { Some(crate::ErrorKind::PackSyncChanged) => { - let mut queue = state.pack_sync_worker.queue.lock(); + let mut queue = state.synced_options.packs.queue.lock(); queue.instances.insert(instance_id.clone()); if !pending.is_empty() { queue @@ -154,7 +162,8 @@ fn start(state: Arc) { let retry_at = Instant::now() + (*retry_in).max(Duration::from_secs(1)); { - let mut queue = state.pack_sync_worker.queue.lock(); + let mut queue = + state.synced_options.packs.queue.lock(); queue.instances.insert(instance_id.clone()); queue .retry_at @@ -170,7 +179,8 @@ fn start(state: Arc) { tokio::spawn(async move { tokio::time::sleep_until(retry_at).await; state - .pack_sync_worker + .synced_options + .packs .queue .lock() .retry_at @@ -205,7 +215,7 @@ impl<'a> Preparation<'a> { Self { state, metadata, - guard: Some(state.lock_synced_options().await), + guard: Some(state.synced_options.lock().await), } } @@ -215,15 +225,24 @@ impl<'a> Preparation<'a> { work: impl Future>, ) -> crate::Result { write_library(library, self.state).await?; - let revision = - self.state.pack_sync_worker.revision.load(Ordering::Acquire); + let revision = self + .state + .synced_options + .packs + .revision + .load(Ordering::Acquire); self.validate_metadata().await?; let before = fingerprint(self.metadata, self.state).await?; self.guard.take(); let result = work.await; - self.guard = Some(self.state.lock_synced_options().await); + self.guard = Some(self.state.synced_options.lock().await); self.validate_metadata().await?; - if self.state.pack_sync_worker.revision.load(Ordering::Acquire) + if self + .state + .synced_options + .packs + .revision + .load(Ordering::Acquire) != revision || fingerprint(self.metadata, self.state).await? != before { diff --git a/packages/app-lib/src/api/instance/synced_servers/modpack.rs b/packages/app-lib/src/api/instance/synced_servers/modpack.rs index 4eb268d7b3..161fb4e8cf 100644 --- a/packages/app-lib/src/api/instance/synced_servers/modpack.rs +++ b/packages/app-lib/src/api/instance/synced_servers/modpack.rs @@ -22,7 +22,7 @@ pub(crate) async fn capture_modpack_server_override( source_path: Option<&str>, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; diff --git a/packages/app-lib/src/api/instance/synced_servers/operations.rs b/packages/app-lib/src/api/instance/synced_servers/operations.rs index 43a856b1b4..0c3bac99cc 100644 --- a/packages/app-lib/src/api/instance/synced_servers/operations.rs +++ b/packages/app-lib/src/api/instance/synced_servers/operations.rs @@ -434,7 +434,7 @@ pub(crate) async fn ensure_managed_server( data: NbtCompound, state: &State, ) -> crate::Result<()> { - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let path = instance_dir(metadata, state).join(SERVERS_FILE); let syncing = participating(metadata, state).await?; let records = list_server_records_locked(metadata, state).await?; @@ -526,7 +526,7 @@ pub(crate) async fn add_user_server( mut data: NbtCompound, state: &State, ) -> crate::Result { - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let id = Uuid::new_v4().to_string(); if participating(metadata, state).await? { let mut canonical = read_canonical(state).await?; @@ -596,7 +596,7 @@ pub(crate) async fn update_server_by_index( accept_textures: Option, state: &State, ) -> crate::Result<()> { - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; if participating(metadata, state).await? { let records = list_server_records_locked(metadata, state).await?; let record = records @@ -655,7 +655,7 @@ pub(crate) async fn remove_server_by_index( index: usize, state: &State, ) -> crate::Result<()> { - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; if participating(metadata, state).await? { let records = list_server_records_locked(metadata, state).await?; let record = records @@ -683,7 +683,7 @@ pub async fn desync_server( mode: DesyncServerMode, ) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let metadata = crate::state::get_instance(instance_id, &state.pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; @@ -735,7 +735,7 @@ pub async fn desync_server( pub async fn list_synced_servers() -> crate::Result> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; Ok(read_canonical(&state) .await? .into_iter() @@ -759,7 +759,7 @@ pub async fn list_synced_servers() -> crate::Result> { pub async fn update_synced_server(server: SyncedServer) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let mut canonical = read_canonical(&state).await?; let target = canonical .iter_mut() @@ -777,7 +777,7 @@ pub async fn update_synced_server(server: SyncedServer) -> crate::Result<()> { pub async fn remove_synced_server(server_id: &str) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let mut canonical = read_canonical(&state).await?; let previous_len = canonical.len(); canonical.retain(|server| server.id != server_id); diff --git a/packages/app-lib/src/api/mr_auth.rs b/packages/app-lib/src/api/mr_auth.rs index a61ee19961..f7c756869a 100644 --- a/packages/app-lib/src/api/mr_auth.rs +++ b/packages/app-lib/src/api/mr_auth.rs @@ -64,7 +64,7 @@ pub async fn authenticate_finish_flow( pub async fn logout() -> crate::Result<()> { let state = crate::State::get().await?; ModrinthCredentials::deactivate_all(&state.pool).await?; - state.friends_socket.disconnect().await?; + state.presence.friends_socket.disconnect().await?; Ok(()) } @@ -99,17 +99,18 @@ pub async fn remove_user(user_id: &str) -> crate::Result<()> { ModrinthCredentials::remove(user_id, &state.pool).await?; if current.is_some_and(|creds| creds.user_id == user_id) { - state.friends_socket.disconnect().await?; + state.presence.friends_socket.disconnect().await?; } Ok(()) } async fn reconnect_friends(state: &crate::State) -> crate::Result<()> { - if let Err(error) = state.friends_socket.disconnect().await { + if let Err(error) = state.presence.friends_socket.disconnect().await { tracing::warn!("Failed to disconnect friends socket: {error}"); } if let Err(error) = state + .presence .friends_socket .connect(&state.pool, &state.api_semaphore, &state.process_manager) .await @@ -127,7 +128,7 @@ pub async fn get_credentials() -> crate::Result> { ModrinthCredentials::get_and_refresh(&state.pool, &state.api_semaphore) .await?; if current.is_none() { - state.friends_socket.disconnect().await?; + state.presence.friends_socket.disconnect().await?; } Ok(current) diff --git a/packages/app-lib/src/api/pack/install_mrpack.rs b/packages/app-lib/src/api/pack/install_mrpack.rs index dea5a6326e..53a0f2b70a 100644 --- a/packages/app-lib/src/api/pack/install_mrpack.rs +++ b/packages/app-lib/src/api/pack/install_mrpack.rs @@ -935,7 +935,8 @@ async fn install_zipped_mrpack_files_with_reporter_inner( project.hashes.get(&PackFileHash::Sha1).and_then( |hash| content_context.file_infos_by_hash.get(hash), ); - let stored_file = file.store_file(state).await?; + let stored_file = + file.store_file(&state.content_store).await?; content_context .reporter .preserve_failure_context( @@ -1184,7 +1185,7 @@ async fn install_zipped_mrpack_files_with_reporter_inner( crate::state::file_modified_at_ns(&io::metadata(&path).await?)?; { - let _permit = state.install_db_semaphore.acquire().await?; + let _permit = state.installs.db_semaphore.acquire().await?; let record_context = InstallErrorContext::new("record modpack override") .maybe_project_id(project_id.clone()) diff --git a/packages/app-lib/src/api/settings.rs b/packages/app-lib/src/api/settings.rs index a9df88a030..f814f9ff75 100644 --- a/packages/app-lib/src/api/settings.rs +++ b/packages/app-lib/src/api/settings.rs @@ -65,7 +65,7 @@ pub async fn store_verify_with_progress( #[tracing::instrument] pub async fn get() -> crate::Result { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let mut settings = Settings::get(&state.pool).await?; if crate::state::game_options_sync_is_enabled(&state.pool).await? && let Some(fullscreen) = @@ -82,7 +82,7 @@ pub async fn get() -> crate::Result { #[tracing::instrument] pub async fn set(settings: Settings) -> crate::Result<()> { let state = State::get().await?; - let _guard = state.lock_synced_options().await; + let _guard = state.synced_options.lock().await; let current_settings = Settings::get(&state.pool).await?; let game_options_sync_enabled = crate::state::game_options_sync_is_enabled(&state.pool).await?; diff --git a/packages/app-lib/src/install/diagnostics.rs b/packages/app-lib/src/install/diagnostics.rs index 776340415f..642a6840b8 100644 --- a/packages/app-lib/src/install/diagnostics.rs +++ b/packages/app-lib/src/install/diagnostics.rs @@ -4,9 +4,9 @@ use super::model::{ InstallPhaseDetails, InstallPhaseId, InstallProgress, }; use super::store; -use crate::state::{ModrinthCredentials, State}; +use crate::state::{DirectoryInfo, ModrinthCredentials, State}; use regex::{Captures, Regex}; -use sqlx::Row; +use sqlx::{Row, SqlitePool}; use std::fmt::Write as _; use std::io::{Read, Seek, SeekFrom}; use std::net::{Ipv4Addr, Ipv6Addr}; @@ -47,9 +47,9 @@ pub async fn build_job_support_details( write_content_summary(&mut details, &job.state.events); write_errors(&mut details, &snapshot); write_raw_snapshot(&mut details, &snapshot); - write_latest_log(&mut details, state).await; + write_latest_log(&mut details, &state.directories).await; - censor_support_text(details, state).await + censor_support_text(details, &state.pool).await } fn result_summary( @@ -411,10 +411,10 @@ fn write_raw_snapshot(details: &mut String, snapshot: &InstallJobSnapshot) { } } -async fn write_latest_log(details: &mut String, state: &State) { +async fn write_latest_log(details: &mut String, directories: &DirectoryInfo) { let _ = writeln!(details); let _ = writeln!(details, "Latest launcher log excerpt"); - match latest_launcher_log_tail(state).await { + match latest_launcher_log_tail(directories).await { Ok(Some((path, output))) => { let _ = writeln!(details, "File: {}", path.display()); details.push_str(&output); @@ -540,9 +540,9 @@ fn json_string(value: &T) -> String { } async fn latest_launcher_log_tail( - state: &State, + directories: &DirectoryInfo, ) -> crate::Result> { - let Some(logs_dir) = state.directories.launcher_logs_dir() else { + let Some(logs_dir) = directories.launcher_logs_dir() else { return Ok(None); }; @@ -600,9 +600,9 @@ fn read_file_tail(path: &Path, max_bytes: u64) -> crate::Result { async fn censor_support_text( mut text: String, - state: &State, + pool: &SqlitePool, ) -> crate::Result { - for credentials in ModrinthCredentials::get_all(&state.pool).await? { + for credentials in ModrinthCredentials::get_all(pool).await? { replace_nonempty( &mut text, &credentials.session, @@ -610,7 +610,7 @@ async fn censor_support_text( ); } - for token in minecraft_tokens(&state.pool).await? { + for token in minecraft_tokens(pool).await? { replace_nonempty(&mut text, &token, "{MINECRAFT_TOKEN}"); } diff --git a/packages/app-lib/src/install/recovery.rs b/packages/app-lib/src/install/recovery.rs index 84146f1b53..d1137a57f8 100644 --- a/packages/app-lib/src/install/recovery.rs +++ b/packages/app-lib/src/install/recovery.rs @@ -12,14 +12,16 @@ use crate::state::content_store::ContentStore; use crate::state::instances::adapters::sqlite::{content_rows, instance_rows}; use crate::state::{ ContentEntry, ContentSetRemoteRef, ContentSetRemoteRefType, - ContentSetSyncProvider, ContentSetSyncState, InstanceFile, + ContentSetSyncProvider, ContentSetSyncState, DirectoryInfo, InstanceFile, InstanceMetadata, State, }; +use crate::util::fetch::IoSemaphore; use anyhow::{Context, bail}; use async_walkdir::WalkDir; use chrono::Utc; use futures::StreamExt; use serde::{Deserialize, Serialize}; +use sqlx::SqlitePool; use std::future::Future; use std::path::{Path, PathBuf}; use uuid::Uuid; @@ -50,8 +52,10 @@ pub(super) async fn prepare_instance_update_backup( state: &State, ) -> crate::Result { let _lease = state.content_store.lease().await; - let _content_lock = - state.lock_instance_content(&metadata.instance.id).await; + let _content_lock = state + .instance_locks + .lock_content(&metadata.instance.id) + .await; let _store_lock = state.content_store.files_lock.lock().await; if crate::state::instance_has_running_process(&metadata.instance.id, state) .await? @@ -61,7 +65,7 @@ pub(super) async fn prepare_instance_update_backup( )); } let owner = job_id.to_string(); - let staging_dir = instance_update_backup_dir(job_id, state); + let staging_dir = instance_update_backup_dir(job_id, &state.directories); if tokio::fs::try_exists(&staging_dir).await? { crate::util::io::remove_dir_all(&staging_dir).await?; } @@ -148,14 +152,19 @@ pub(super) async fn prepare_instance_update_backup( &instance_path.join(path), &backup_path.join(path), &skipped, - state, + &state.io_semaphore, ) .await?; } } None => { - copy_directory(&instance_path, &backup_path, &skipped, state) - .await?; + copy_directory( + &instance_path, + &backup_path, + &skipped, + &state.io_semaphore, + ) + .await?; } } let snapshot = SharedInstanceUpdateRollback { @@ -188,11 +197,11 @@ pub(super) async fn prepare_instance_update_backup( Ok(staging_dir) } -fn instance_update_backup_dir(job_id: Uuid, state: &State) -> PathBuf { - state - .directories - .install_backups_dir() - .join(job_id.to_string()) +fn instance_update_backup_dir( + job_id: Uuid, + directories: &DirectoryInfo, +) -> PathBuf { + directories.install_backups_dir().join(job_id.to_string()) } async fn recover_unrecorded_instance_update_backup( @@ -209,7 +218,7 @@ async fn recover_unrecorded_instance_update_backup( { return Ok(()); } - let staging_dir = instance_update_backup_dir(job.id, state); + let staging_dir = instance_update_backup_dir(job.id, &state.directories); if !tokio::fs::try_exists(&staging_dir).await? { return Ok(()); } @@ -267,7 +276,7 @@ async fn restore_instance_update( state: &State, ) -> crate::Result<()> { let instance_id = &rollback.instance.instance.id; - let _content_lock = state.lock_instance_content(instance_id).await; + let _content_lock = state.instance_locks.lock_content(instance_id).await; let _store_lock = state.content_store.files_lock.lock().await; let _lease = state.content_store.lease().await; if crate::state::instance_has_running_process(instance_id, state).await? { @@ -334,7 +343,7 @@ async fn restore_instance_update( &backup_path.join(path), &target, &std::collections::HashSet::new(), - state, + &state.io_semaphore, ) .await?; } @@ -347,7 +356,7 @@ async fn restore_instance_update( &backup_path, &instance_path, &std::collections::HashSet::new(), - state, + &state.io_semaphore, ) .await?; } @@ -369,17 +378,17 @@ async fn restore_instance_update( &snapshot.copied_file_ids, ) .await?; - restore_instance_metadata(&rollback.instance, state).await?; + restore_instance_metadata(&rollback.instance, &state.pool).await?; Ok(()) } async fn restore_instance_metadata( metadata: &InstanceMetadata, - state: &State, + pool: &SqlitePool, ) -> crate::Result<()> { let content_set_id = metadata.applied_content_set.id.as_str(); - let mut tx = state.pool.begin_with("BEGIN IMMEDIATE").await?; + let mut tx = pool.begin_with("BEGIN IMMEDIATE").await?; instance_rows::update_instance(&metadata.instance, &mut tx).await?; content_rows::update_content_set(&metadata.applied_content_set, &mut tx) .await?; @@ -493,18 +502,16 @@ async fn copy_path( source: &Path, target: &Path, skipped: &std::collections::HashSet, - state: &State, + io_semaphore: &IoSemaphore, ) -> crate::Result<()> { match symlink_metadata(source).await? { Some(metadata) if metadata.is_symlink() => { copy_symlink(source, target).await } Some(metadata) if metadata.is_dir() => { - copy_directory(source, target, skipped, state).await - } - Some(_) => { - crate::util::fetch::copy(source, target, &state.io_semaphore).await + copy_directory(source, target, skipped, io_semaphore).await } + Some(_) => crate::util::fetch::copy(source, target, io_semaphore).await, None => Ok(()), } } @@ -513,7 +520,7 @@ async fn copy_directory( source: &Path, target: &Path, skipped: &std::collections::HashSet, - state: &State, + io_semaphore: &IoSemaphore, ) -> crate::Result<()> { crate::util::io::create_dir_all(target).await?; let mut walker = WalkDir::new(source); @@ -536,12 +543,8 @@ async fn copy_directory( if file_type.is_dir() { crate::util::io::create_dir_all(&target_path).await?; } else if file_type.is_file() { - crate::util::fetch::copy( - &entry_path, - &target_path, - &state.io_semaphore, - ) - .await?; + crate::util::fetch::copy(&entry_path, &target_path, io_semaphore) + .await?; } else if file_type.is_symlink() { copy_symlink(&entry_path, &target_path).await?; } diff --git a/packages/app-lib/src/install/runner.rs b/packages/app-lib/src/install/runner.rs index 2eb11972a5..271ad9d212 100644 --- a/packages/app-lib/src/install/runner.rs +++ b/packages/app-lib/src/install/runner.rs @@ -49,9 +49,6 @@ const SYNCED_OPTION_FILES: [&str; 4] = [ "command_history.txt", ]; -/// Admission covers setup and deletion. A target reservation stays with its worker -/// until cleanup finishes, so backups and rollback cannot overlap another install. -static INSTALL_ADMISSION: AsyncMutex<()> = AsyncMutex::const_new(()); static INSTALL_TARGETS: LazyLock>>>> = LazyLock::new(|| Mutex::new(HashMap::new())); @@ -243,8 +240,8 @@ pub fn retry_job( } async fn retry_job_inner(job_id: Uuid) -> crate::Result { - let _admission = INSTALL_ADMISSION.lock().await; let state = State::get().await?; + let _admission = state.installs.admission.lock().await; let mut job = store::get_required(job_id, &state.pool).await?; if !matches!( @@ -427,11 +424,11 @@ pub async fn cancel_job(job_id: Uuid) -> crate::Result { /// The caller must retain both guards until the instance has been removed. /// This prevents a new install from starting after cancellation has finished. -pub(crate) async fn cancel_jobs_for_instance_deletion( +pub(crate) async fn cancel_jobs_for_instance_deletion<'a>( instance_id: &str, - state: &State, -) -> crate::Result<(MutexGuard<'static, ()>, OwnedMutexGuard<()>)> { - let admission = INSTALL_ADMISSION.lock().await; + state: &'a State, +) -> crate::Result<(MutexGuard<'a, ()>, OwnedMutexGuard<()>)> { + let admission = state.installs.admission.lock().await; let jobs = store::list_active_for_instance(instance_id, &state.pool).await?; for job in &jobs { @@ -489,9 +486,9 @@ fn start( async fn start_inner( request: InstallRequest, ) -> crate::Result { - let _admission = INSTALL_ADMISSION.lock().await; - let mut target_guard = reserve_target(&request.target())?; let state = State::get().await?; + let _admission = state.installs.admission.lock().await; + let mut target_guard = reserve_target(&request.target())?; if let InstallTarget::ExistingInstance { instance_id } = request.target() { store::ensure_no_pending_recovery(&instance_id, None, &state.pool) .await?; @@ -692,7 +689,8 @@ async fn prepare_initial_instance_inner( ); let instance_id = metadata.instance.id; set_instance_id(job_state, instance_id.clone()); - attach_pending_shared_instance(&instance_id, &data, state).await?; + attach_pending_shared_instance(&instance_id, &data, &state.pool) + .await?; emit_instance(&instance_id, InstancePayloadType::Edited).await?; } InstallRequest::ImportInstance { @@ -745,7 +743,8 @@ async fn prepare_initial_instance_inner( } | InstallRequest::BulkUpdateContent { instance_id, .. } | InstallRequest::UpdateSharedInstance { instance_id, .. } => { - prepare_existing_rollback(job_state, state, &instance_id).await?; + prepare_existing_rollback(job_state, &state.pool, &instance_id) + .await?; } } @@ -796,7 +795,7 @@ async fn run_job_inner( let _install_permit = if control.checkpoint().await.is_ok() { tokio::select! { - permit = state.install_job_semaphore.acquire() => Some(permit?), + permit = state.installs.job_semaphore.acquire() => Some(permit?), () = control.canceled() => None, } } else { @@ -1065,7 +1064,7 @@ async fn run_request_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::PreparingInstance, InstallPhaseDetails::Instance { name: name.clone() }, ) @@ -1073,7 +1072,7 @@ async fn run_request_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::DownloadingMinecraft, InstallPhaseDetails::Minecraft { game_version, @@ -1113,7 +1112,7 @@ async fn run_request_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::ResolvingPack, modpack_details(&location), ) @@ -1146,8 +1145,12 @@ async fn run_request_inner( ) .await?; - finalize_shared_instance_attachment(&instance_id, &data, state) - .await?; + finalize_shared_instance_attachment( + &instance_id, + &data, + &state.pool, + ) + .await?; emit_instance(&instance_id, InstancePayloadType::Edited).await?; Ok(Some(instance_id)) @@ -1165,7 +1168,7 @@ async fn run_request_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::PreparingInstance, InstallPhaseDetails::Import { launcher_type, @@ -1196,7 +1199,7 @@ async fn run_request_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::PreparingInstance, InstallPhaseDetails::Empty, ) @@ -1236,12 +1239,13 @@ async fn run_request_inner( Ok(Some(instance_id)) } InstallRequest::InstallExistingInstance { instance_id, force } => { - prepare_existing_rollback(job_state, state, &instance_id).await?; + prepare_existing_rollback(job_state, &state.pool, &instance_id) + .await?; lock_instance(&instance_id, state).await?; update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::DownloadingMinecraft, InstallPhaseDetails::Empty, ) @@ -1271,7 +1275,8 @@ async fn run_request_inner( location, post_install_edit, } => { - prepare_existing_rollback(job_state, state, &instance_id).await?; + prepare_existing_rollback(job_state, &state.pool, &instance_id) + .await?; lock_instance(&instance_id, state).await?; prepare_update_backup(job_id, job_state, state).await?; crate::api::instance::prepare_instance_update(&instance_id).await?; @@ -1309,7 +1314,7 @@ async fn run_request_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::ResolvingPack, InstallPhaseDetails::Empty, ) @@ -1336,7 +1341,8 @@ async fn run_request_inner( Ok(Some(instance_id)) } InstallRequest::UpdateSharedInstance { instance_id, data } => { - prepare_existing_rollback(job_state, state, &instance_id).await?; + prepare_existing_rollback(job_state, &state.pool, &instance_id) + .await?; lock_instance(&instance_id, state).await?; prepare_update_backup(job_id, job_state, state).await?; let disabled_project_ids = @@ -1355,8 +1361,12 @@ async fn run_request_inner( state, ) .await?; - finalize_shared_instance_attachment(&instance_id, &data, state) - .await?; + finalize_shared_instance_attachment( + &instance_id, + &data, + &state.pool, + ) + .await?; emit_instance(&instance_id, InstancePayloadType::Edited).await?; Ok(Some(instance_id)) } @@ -1639,18 +1649,16 @@ async fn install_pack_inner( async fn prepare_existing_rollback( job_state: &mut InstallJobState, - state: &State, + pool: &SqlitePool, instance_id: &str, ) -> crate::Result<()> { if job_state.rollback.is_some() { return Ok(()); } - let instance = crate::state::get_instance(instance_id, &state.pool) - .await? - .ok_or(ErrorKind::InputError(format!( - "unknown instance {instance_id}" - )))?; + let instance = crate::state::get_instance(instance_id, pool).await?.ok_or( + ErrorKind::InputError(format!("unknown instance {instance_id}")), + )?; if instance.quarantined { bail!(ErrorKind::InputError( "content in quarantined instances cannot be changed".to_string(), @@ -1841,7 +1849,7 @@ async fn lock_install_target( } async fn lock_instance(instance_id: &str, state: &State) -> crate::Result<()> { - let _content_lock = state.lock_instance_content(instance_id).await; + let _content_lock = state.instance_locks.lock_content(instance_id).await; if crate::state::instance_has_running_process(instance_id, state).await? { bail!(ErrorKind::InputError( "stop this instance before installing or updating its content" @@ -1862,13 +1870,13 @@ async fn lock_instance(instance_id: &str, state: &State) -> crate::Result<()> { pub(super) async fn update_progress( job_id: Uuid, job_state: &mut InstallJobState, - state: &State, + pool: &SqlitePool, phase: InstallPhaseId, details: InstallPhaseDetails, ) -> crate::Result<()> { super::control::checkpoint(job_id).await?; job_state.set_progress(phase, None, details); - let record = store::update_state(job_id, job_state, &state.pool).await?; + let record = store::update_state(job_id, job_state, pool).await?; emit_install_job(&record.snapshot()).await?; Ok(()) } @@ -1876,7 +1884,7 @@ pub(super) async fn update_progress( pub(super) async fn update_content_progress( job_id: Uuid, job_state: &mut InstallJobState, - state: &State, + pool: &SqlitePool, current: u64, total: u64, ) -> crate::Result<()> { @@ -1888,7 +1896,7 @@ pub(super) async fn update_content_progress( secondary: None, }); job_state.progress.details = InstallPhaseDetails::Empty; - let record = store::update_state(job_id, job_state, &state.pool).await?; + let record = store::update_state(job_id, job_state, pool).await?; emit_install_job(&record.snapshot()).await?; Ok(()) } diff --git a/packages/app-lib/src/install/shared_instance.rs b/packages/app-lib/src/install/shared_instance.rs index 8b5d09680b..1362b637cc 100644 --- a/packages/app-lib/src/install/shared_instance.rs +++ b/packages/app-lib/src/install/shared_instance.rs @@ -23,6 +23,7 @@ use crate::util::fetch::{DownloadReason, REQWEST_CLIENT}; use anyhow::{Context, bail}; use futures::StreamExt; use path_util::SafeRelativeUtf8UnixPathBuf; +use sqlx::SqlitePool; use std::collections::{HashMap, HashSet}; use std::future::Future; use std::path::PathBuf; @@ -34,7 +35,7 @@ const MAX_SHARED_INSTANCE_INITIAL_BUFFER_SIZE: u64 = 8 * 1024 * 1024; pub(super) async fn attach_pending_shared_instance( instance_id: &str, data: &SharedInstanceInstallData, - state: &State, + pool: &SqlitePool, ) -> crate::Result<()> { crate::state::attach_shared_instance( instance_id, @@ -49,7 +50,7 @@ pub(super) async fn attach_pending_shared_instance( applied_version: None, latest_version: Some(data.version), }, - &state.pool, + pool, ) .await } @@ -57,7 +58,7 @@ pub(super) async fn attach_pending_shared_instance( pub(super) async fn finalize_shared_instance_attachment( instance_id: &str, data: &SharedInstanceInstallData, - state: &State, + pool: &SqlitePool, ) -> crate::Result<()> { crate::state::attach_shared_instance( instance_id, @@ -72,7 +73,7 @@ pub(super) async fn finalize_shared_instance_attachment( applied_version: Some(data.version), latest_version: Some(data.version), }, - &state.pool, + pool, ) .await } @@ -296,7 +297,7 @@ async fn apply_shared_instance_update_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::PreparingInstance, InstallPhaseDetails::Instance { name: data.name.clone(), @@ -309,7 +310,7 @@ async fn apply_shared_instance_update_inner( update_content_progress( job_id, job_state, - state, + &state.pool, 0, content_change_count, ) @@ -359,7 +360,7 @@ async fn apply_shared_instance_update_inner( update_content_progress( job_id, job_state, - state, + &state.pool, completed_content_changes, content_change_count, ) @@ -380,7 +381,7 @@ async fn apply_shared_instance_update_inner( update_content_progress( job_id, job_state, - state, + &state.pool, completed_content_changes, content_change_count, ) @@ -398,7 +399,7 @@ async fn apply_shared_instance_update_inner( update_content_progress( job_id, job_state, - state, + &state.pool, completed_content_changes, content_change_count, ) @@ -416,7 +417,7 @@ async fn apply_shared_instance_update_inner( update_content_progress( job_id, job_state, - state, + &state.pool, completed_content_changes, content_change_count, ) @@ -639,7 +640,7 @@ async fn apply_shared_instance_content_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::PreparingInstance, InstallPhaseDetails::Instance { name: data.name.clone(), @@ -660,7 +661,7 @@ async fn apply_shared_instance_content_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::ResolvingPack, modpack_details(&location), ) @@ -692,7 +693,7 @@ async fn apply_shared_instance_content_inner( update_progress( job_id, job_state, - state, + &state.pool, InstallPhaseId::DownloadingMinecraft, InstallPhaseDetails::Minecraft { game_version: data.game_version.clone(), @@ -726,7 +727,7 @@ async fn apply_shared_instance_content_inner( update_content_progress( job_id, job_state, - state, + &state.pool, 0, content_change_count, ) @@ -746,7 +747,7 @@ async fn apply_shared_instance_content_inner( update_content_progress( job_id, job_state, - state, + &state.pool, completed_content_changes, content_change_count, ) @@ -760,7 +761,7 @@ async fn apply_shared_instance_content_inner( update_content_progress( job_id, job_state, - state, + &state.pool, completed_content_changes, content_change_count, ) diff --git a/packages/app-lib/src/launcher/mod.rs b/packages/app-lib/src/launcher/mod.rs index a779b60fe7..b5e7180e51 100644 --- a/packages/app-lib/src/launcher/mod.rs +++ b/packages/app-lib/src/launcher/mod.rs @@ -1216,7 +1216,7 @@ pub async fn launch_minecraft( .await .context("syncing instance content files")?; let _instance_content_lock = - state.lock_instance_content(&instance.id).await; + state.instance_locks.lock_content(&instance.id).await; let current = crate::state::instances::adapters::sqlite::instance_rows::get_instance_by_id( &instance.id, &state.pool, @@ -1271,11 +1271,13 @@ pub async fn launch_minecraft( } let _ = state + .presence .discord_rpc .set_activity(&format!("Playing {}", instance.name), true, &state.pool) .await; let _ = state + .presence .friends_socket .update_status(Some(instance.name.clone())) .await; diff --git a/packages/app-lib/src/state/content_store/adapters/downloads.rs b/packages/app-lib/src/state/content_store/adapters/downloads.rs index 54c3035826..c81b0119b4 100644 --- a/packages/app-lib/src/state/content_store/adapters/downloads.rs +++ b/packages/app-lib/src/state/content_store/adapters/downloads.rs @@ -28,7 +28,7 @@ pub(in crate::state::content_store) async fn download( pub(in crate::state::content_store) async fn repair_sources( pool: &SqlitePool, sha512: &str, - state: &crate::State, + api_semaphore: &FetchSemaphore, ) -> Vec { let version = fetch::fetch_json::( reqwest::Method::GET, @@ -39,7 +39,7 @@ pub(in crate::state::content_store) async fn repair_sources( None, None, Some("/v2/version_file/:hash"), - &state.api_semaphore, + api_semaphore, pool, ) .await; diff --git a/packages/app-lib/src/state/content_store/commands/migration.rs b/packages/app-lib/src/state/content_store/commands/migration.rs index 24bdd90efa..a5e2436a07 100644 --- a/packages/app-lib/src/state/content_store/commands/migration.rs +++ b/packages/app-lib/src/state/content_store/commands/migration.rs @@ -209,7 +209,7 @@ async fn migrate_file_copy( let copy_ms = started.elapsed().as_millis() as u64; let waiting = Instant::now(); let _turn = store.legacy_migration_priority.read().await; - let _instance = state.lock_instance_content(&instance.id).await; + let _instance = state.instance_locks.lock_content(&instance.id).await; let _files = store.files_lock.lock().await; let lock_wait_ms = waiting.elapsed().as_millis() as u64; let Some(instance) = @@ -285,7 +285,7 @@ async fn migrate_file_copy( sources = downloads::repair_sources( &state.pool, &stored.sha512, - state, + &state.api_semaphore, ) .await; } diff --git a/packages/app-lib/src/state/content_store/commands/verify_content.rs b/packages/app-lib/src/state/content_store/commands/verify_content.rs index 34a469dd6a..4c00133339 100644 --- a/packages/app-lib/src/state/content_store/commands/verify_content.rs +++ b/packages/app-lib/src/state/content_store/commands/verify_content.rs @@ -195,7 +195,7 @@ impl ContentStore { sources = downloads::repair_sources( &self.pool, &stored_file.sha512, - state, + &state.api_semaphore, ) .await; } diff --git a/packages/app-lib/src/state/friends.rs b/packages/app-lib/src/state/friends.rs index f6c15d96a5..55447ffa99 100644 --- a/packages/app-lib/src/state/friends.rs +++ b/packages/app-lib/src/state/friends.rs @@ -297,7 +297,8 @@ impl FriendsSocket { let mut last_ping = Utc::now(); loop { - let connected = state.friends_socket.is_connected().await; + let connected = + state.presence.friends_socket.is_connected().await; if !connected && Utc::now().signed_duration_since(last_connection) @@ -306,6 +307,7 @@ impl FriendsSocket { last_connection = Utc::now(); last_ping = Utc::now(); let _ = state + .presence .friends_socket .connect( &state.pool, @@ -318,7 +320,8 @@ impl FriendsSocket { > chrono::Duration::seconds(10) { last_ping = Utc::now(); - let mut write = state.friends_socket.write.write().await; + let mut write = + state.presence.friends_socket.write.write().await; if let Some(write) = write.as_mut() { let _ = write.send(Message::Ping(Bytes::new())).await; } diff --git a/packages/app-lib/src/state/installs.rs b/packages/app-lib/src/state/installs.rs new file mode 100644 index 0000000000..349f88fe24 --- /dev/null +++ b/packages/app-lib/src/state/installs.rs @@ -0,0 +1,22 @@ +use tokio::sync::{Mutex, Semaphore}; + +const MAX_CONCURRENT_INSTALL_JOBS: usize = 3; + +/// Limits install job concurrency. +pub(crate) struct Installs { + /// Admission covers setup and deletion. A target reservation stays with its worker + /// until cleanup finishes, so backups and rollback cannot overlap another install. + pub(crate) admission: Mutex<()>, + pub(crate) job_semaphore: Semaphore, + pub(crate) db_semaphore: Semaphore, +} + +impl Installs { + pub(crate) fn new() -> Self { + Self { + admission: Mutex::new(()), + job_semaphore: Semaphore::new(MAX_CONCURRENT_INSTALL_JOBS), + db_semaphore: Semaphore::new(1), + } + } +} diff --git a/packages/app-lib/src/state/instance_locks.rs b/packages/app-lib/src/state/instance_locks.rs new file mode 100644 index 0000000000..3508f5f8d3 --- /dev/null +++ b/packages/app-lib/src/state/instance_locks.rs @@ -0,0 +1,54 @@ +use dashmap::DashMap; +use std::sync::Arc; +use tokio::sync::{Mutex, OwnedMutexGuard}; + +/// Serializes concurrent changes per instance. +#[derive(Default)] +pub(crate) struct InstanceLocks { + /// Serializes filesystem reconciliation and content mutations per instance. + content: DashMap>>, + screenshots: DashMap>>, + /// Serializes shared instance attachment and recipient mutations per instance. + shared: DashMap>>, +} + +impl InstanceLocks { + pub(crate) async fn lock_content( + &self, + instance_id: &str, + ) -> OwnedMutexGuard<()> { + lock(&self.content, instance_id).await + } + + pub(crate) async fn lock_screenshots( + &self, + instance_id: &str, + ) -> OwnedMutexGuard<()> { + lock(&self.screenshots, instance_id).await + } + + pub(crate) async fn lock_shared( + &self, + instance_id: &str, + ) -> OwnedMutexGuard<()> { + lock(&self.shared, instance_id).await + } + + pub(crate) fn remove(&self, instance_id: &str) { + let _ = self.content.remove(instance_id); + let _ = self.screenshots.remove(instance_id); + let _ = self.shared.remove(instance_id); + } +} + +async fn lock( + locks: &DashMap>>, + instance_id: &str, +) -> OwnedMutexGuard<()> { + let lock = locks + .entry(instance_id.to_string()) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone(); + + lock.lock_owned().await +} diff --git a/packages/app-lib/src/state/instances/commands/apply_content_install.rs b/packages/app-lib/src/state/instances/commands/apply_content_install.rs index e2dd5cea9b..29ab8e3330 100644 --- a/packages/app-lib/src/state/instances/commands/apply_content_install.rs +++ b/packages/app-lib/src/state/instances/commands/apply_content_install.rs @@ -4,8 +4,8 @@ use crate::state::instances::{ adapters::sqlite::{content_rows, instance_rows}, }; use crate::state::{ - CacheBehaviour, CachedEntry, Dependency, DependencyType, ModLoader, - ProjectType, State, Version, + CacheBehaviour, CachedEntry, Dependency, DependencyType, DirectoryInfo, + ModLoader, ProjectType, State, Version, }; use crate::util::fetch::{self, DownloadMeta, DownloadReason}; use crate::util::io; @@ -17,6 +17,7 @@ use modrinth_content_management::{ ResolutionPreferences, ResolveContentPlan, ResolveContentRequest, ResolvedContent, }; +use sqlx::SqlitePool; use std::path::{Path, PathBuf}; use super::content_mutation::{ @@ -323,9 +324,9 @@ async fn add_resolved_content( pub(crate) async fn resolve_content_scope( instance_id: &str, content_set_id: Option<&str>, - state: &State, + pool: &SqlitePool, ) -> crate::Result { - let instance = instance_rows::get_instance_by_id(instance_id, &state.pool) + let instance = instance_rows::get_instance_by_id(instance_id, pool) .await? .ok_or(ErrorKind::InputError("unknown instance".to_string()))?; let content_set_id = match content_set_id { @@ -391,7 +392,7 @@ pub(crate) async fn download_project_version_with_progress( state: &State, progress: Option<&mut fetch::FetchProgressFn<'_>>, ) -> crate::Result { - let scope = resolve_content_scope(instance_id, None, state).await?; + let scope = resolve_content_scope(instance_id, None, &state.pool).await?; let content_set = content_rows::get_content_set(&scope.content_set_id, &state.pool) .await? @@ -478,7 +479,7 @@ pub(crate) async fn add_downloaded_project_version_with_enabled( "invalid project filename".to_string() )); } - let stored_file = downloaded.file.store_file(state).await?; + let stored_file = downloaded.file.store_file(&state.content_store).await?; install_stored_file( instance_id, InstallContent { @@ -610,7 +611,7 @@ pub(crate) async fn content_source_kind_for_project_path( project_path: &str, state: &State, ) -> crate::Result> { - let scope = resolve_content_scope(instance_id, None, state).await?; + let scope = resolve_content_scope(instance_id, None, &state.pool).await?; let Some(file) = content_rows::get_instance_file_by_relative_path( &scope.instance.id, project_path, @@ -635,7 +636,7 @@ pub(crate) async fn is_project_locked( project_path: &str, state: &State, ) -> crate::Result { - let scope = resolve_content_scope(instance_id, None, state).await?; + let scope = resolve_content_scope(instance_id, None, &state.pool).await?; content_rows::is_instance_file_locked( &scope.instance.id, project_path, @@ -650,8 +651,8 @@ pub(crate) async fn set_project_locked( locked: bool, state: &State, ) -> crate::Result<()> { - let _content_lock = state.lock_instance_content(instance_id).await; - let scope = resolve_content_scope(instance_id, None, state).await?; + let _content_lock = state.instance_locks.lock_content(instance_id).await; + let scope = resolve_content_scope(instance_id, None, &state.pool).await?; content_rows::set_instance_file_locked( &scope.instance.id, project_path, @@ -669,8 +670,9 @@ pub(crate) async fn rename_project_companion_file( ) -> crate::Result<()> { let project_type = ProjectType::get_from_parent_folder(new_project_path); if project_type == Some(ProjectType::ShaderPack) { - let scope = resolve_content_scope(instance_id, None, state).await?; - let base = instance_full_path(state, &scope.instance); + let scope = + resolve_content_scope(instance_id, None, &state.pool).await?; + let base = instance_full_path(&state.directories, &scope.instance); let old_txt_path = base.join(format!( "{}.txt", @@ -701,7 +703,7 @@ pub(crate) async fn list_project_files( instance_id: &str, state: &State, ) -> crate::Result> { - let scope = resolve_content_scope(instance_id, None, state).await?; + let scope = resolve_content_scope(instance_id, None, &state.pool).await?; let entries = content_rows::get_content_entries(&scope.content_set_id, &state.pool) .await?; @@ -726,10 +728,10 @@ pub(crate) async fn list_project_files( } pub(crate) fn instance_full_path( - state: &State, + directories: &DirectoryInfo, instance: &Instance, ) -> PathBuf { - state.directories.instances_dir().join(&instance.path) + directories.instances_dir().join(&instance.path) } pub(super) async fn upsert_entry_for_file( diff --git a/packages/app-lib/src/state/instances/commands/apply_content_update.rs b/packages/app-lib/src/state/instances/commands/apply_content_update.rs index 565bed3e0e..f890015030 100644 --- a/packages/app-lib/src/state/instances/commands/apply_content_update.rs +++ b/packages/app-lib/src/state/instances/commands/apply_content_update.rs @@ -15,6 +15,7 @@ use crate::state::{ use crate::util::fetch::DownloadReason; use anyhow::bail; use futures::stream::{self, StreamExt}; +use sqlx::SqlitePool; use std::cmp::Reverse; use std::collections::{HashMap, HashSet}; use std::future::Future; @@ -375,7 +376,7 @@ pub(crate) async fn plan_bulk_update( state: &State, ) -> crate::Result { let shared_instance_member = - is_shared_instance_member(instance_id, state).await?; + is_shared_instance_member(instance_id, &state.pool).await?; let updateable_paths = bulk_updateable_project_paths( instance_id, shared_instance_member, @@ -688,11 +689,10 @@ fn installed_project_from_row( async fn is_shared_instance_member( instance_id: &str, - state: &State, + pool: &SqlitePool, ) -> crate::Result { let Some(metadata) = - instance_rows::get_instance_metadata_by_id(instance_id, &state.pool) - .await? + instance_rows::get_instance_metadata_by_id(instance_id, pool).await? else { return Ok(false); }; diff --git a/packages/app-lib/src/state/instances/commands/content_mutation.rs b/packages/app-lib/src/state/instances/commands/content_mutation.rs index e64b3bfc57..9cced0aadd 100644 --- a/packages/app-lib/src/state/instances/commands/content_mutation.rs +++ b/packages/app-lib/src/state/instances/commands/content_mutation.rs @@ -101,7 +101,7 @@ impl<'a> InstanceContent<'a> { instance_id: &str, state: &'a State, ) -> crate::Result { - let content_lock = state.lock_instance_content(instance_id).await; + let content_lock = state.instance_locks.lock_content(instance_id).await; let store_lock = state.content_store.files_lock.lock().await; let instance = instance_rows::get_instance_by_id(instance_id, &state.pool) diff --git a/packages/app-lib/src/state/instances/commands/create_instance.rs b/packages/app-lib/src/state/instances/commands/create_instance.rs index b95f4f5323..43c50277b5 100644 --- a/packages/app-lib/src/state/instances/commands/create_instance.rs +++ b/packages/app-lib/src/state/instances/commands/create_instance.rs @@ -12,6 +12,7 @@ use crate::util::fetch; use crate::util::io; use chrono::Utc; use serde::{Deserialize, Serialize}; +use sqlx::SqlitePool; use tracing::{info, trace}; use uuid::Uuid; @@ -166,7 +167,7 @@ async fn resolve_instance_path( let mut path = base_path.clone(); let mut full_path = state.directories.instances_dir().join(&path); - if path_available(&path, &full_path, state).await? { + if path_available(&path, &full_path, &state.pool).await? { return Ok((path, full_path)); } @@ -175,7 +176,7 @@ async fn resolve_instance_path( path = format!("{base_path} ({which})"); full_path = state.directories.instances_dir().join(&path); - if path_available(&path, &full_path, state).await? { + if path_available(&path, &full_path, &state.pool).await? { return Ok((path, full_path)); } @@ -186,13 +187,13 @@ async fn resolve_instance_path( async fn path_available( path: &str, full_path: &std::path::Path, - state: &State, + pool: &SqlitePool, ) -> crate::Result { if full_path.exists() { return Ok(false); } - Ok(instance_rows::get_instance_by_path(path, &state.pool) + Ok(instance_rows::get_instance_by_path(path, pool) .await? .is_none()) } diff --git a/packages/app-lib/src/state/instances/commands/remove_instance.rs b/packages/app-lib/src/state/instances/commands/remove_instance.rs index ec8195cc88..b6f4511311 100644 --- a/packages/app-lib/src/state/instances/commands/remove_instance.rs +++ b/packages/app-lib/src/state/instances/commands/remove_instance.rs @@ -11,8 +11,8 @@ pub(crate) async fn remove_instance( let instance = instance_rows::get_instance_by_id(instance_id, &state.pool) .await? .ok_or(crate::ErrorKind::InputError("unknown instance".to_string()))?; - let _synced_options_lock = state.lock_synced_options().await; - let _content_lock = state.lock_instance_content(instance_id).await; + let _synced_options_lock = state.synced_options.lock().await; + let _content_lock = state.instance_locks.lock_content(instance_id).await; let _store_lock = state.content_store.files_lock.lock().await; let _store_lease = state.content_store.lease().await; if crate::state::instance_has_running_process(instance_id, state).await? { @@ -40,7 +40,7 @@ async fn delete_instance_row_and_locks( ) -> crate::Result<()> { // Keep these together so deleted instances cannot leave stale entries in the per-instance lock maps. instance_rows::delete_instance_by_id(instance_id, &state.pool).await?; - state.remove_instance_locks(instance_id); + state.instance_locks.remove(instance_id); Ok(()) } diff --git a/packages/app-lib/src/state/mod.rs b/packages/app-lib/src/state/mod.rs index 5759d4799d..341aeabe31 100644 --- a/packages/app-lib/src/state/mod.rs +++ b/packages/app-lib/src/state/mod.rs @@ -2,11 +2,10 @@ use crate::ErrorKind; use crate::util::fetch::{FetchSemaphore, IoSemaphore}; use anyhow::anyhow; -use dashmap::DashMap; use std::sync::atomic::AtomicBool; use std::sync::{Arc, LazyLock}; use tokio::sync::watch::Sender; -use tokio::sync::{Mutex, MutexGuard, OnceCell, OwnedMutexGuard, Semaphore}; +use tokio::sync::{Mutex, OnceCell, Semaphore}; use crate::state::instances::watcher::FileWatcher; use sqlx::SqlitePool; @@ -56,6 +55,18 @@ pub(crate) mod runtime_cache; mod friends; pub use self::friends::*; +mod installs; +pub(crate) use self::installs::Installs; + +mod instance_locks; +pub(crate) use self::instance_locks::InstanceLocks; + +mod presence; +pub use self::presence::Presence; + +mod synced_options; +pub(crate) use self::synced_options::SyncedOptions; + mod tunnel; pub use self::tunnel::*; @@ -80,7 +91,6 @@ static LAUNCHER_STATE: OnceCell> = OnceCell::const_new(); static STATE_STARTUP: LazyLock> = LazyLock::new(|| Sender::new(StartupPhase::Pending)); static STATE_STARTUP_LOCK: Mutex<()> = Mutex::const_new(()); -const MAX_CONCURRENT_INSTALL_JOBS: usize = 3; pub struct State { startup_complete: AtomicBool, /// Information on the location of files used in the launcher @@ -94,21 +104,11 @@ pub struct State { /// Semaphore to limit concurrent API requests. This is separate from the fetch semaphore /// to keep API functionality while the app is performing intensive tasks. pub api_semaphore: FetchSemaphore, - pub(crate) install_job_semaphore: Semaphore, - pub(crate) install_db_semaphore: Semaphore, - /// Serializes filesystem reconciliation and content mutations per instance. - instance_content_locks: DashMap>>, - /// Serializes screenshot filesystem reconciliation per instance. - instance_screenshot_locks: DashMap>>, - /// Serializes shared instance attachment and recipient mutations per instance. - shared_instance_locks: DashMap>>, - /// Serializes canonical synced-option mutations and checkpoint updates. - synced_options_lock: Mutex<()>, - pub(crate) game_locale_indexer: crate::api::instance::GameLocaleIndexer, - pub(crate) pack_sync_worker: crate::api::instance::PackSyncWorker, - - /// Discord RPC - pub discord_rpc: DiscordGuard, + pub(crate) installs: Installs, + pub(crate) instance_locks: InstanceLocks, + pub(crate) synced_options: SyncedOptions, + + pub presence: Presence, /// Process manager pub process_manager: ProcessManager, @@ -119,8 +119,7 @@ pub struct State { // // /// App identifier string (like com.modrinth.ModrinthApp) // pub app_identifier: String, - /// Friends socket - pub friends_socket: FriendsSocket, + pub restart_after_pending_update: AtomicBool, pub(crate) pool: SqlitePool, @@ -128,55 +127,6 @@ pub struct State { } impl State { - pub(crate) async fn lock_synced_options(&self) -> MutexGuard<'_, ()> { - self.synced_options_lock.lock().await - } - - pub(crate) async fn lock_instance_content( - &self, - instance_id: &str, - ) -> OwnedMutexGuard<()> { - let lock = self - .instance_content_locks - .entry(instance_id.to_string()) - .or_insert_with(|| Arc::new(Mutex::new(()))) - .clone(); - - lock.lock_owned().await - } - - pub(crate) async fn lock_shared_instance( - &self, - instance_id: &str, - ) -> OwnedMutexGuard<()> { - let lock = self - .shared_instance_locks - .entry(instance_id.to_string()) - .or_insert_with(|| Arc::new(Mutex::new(()))) - .clone(); - - lock.lock_owned().await - } - - pub(crate) async fn lock_instance_screenshots( - &self, - instance_id: &str, - ) -> OwnedMutexGuard<()> { - let lock = self - .instance_screenshot_locks - .entry(instance_id.to_string()) - .or_insert_with(|| Arc::new(Mutex::new(()))) - .clone(); - - lock.lock_owned().await - } - - pub(crate) fn remove_instance_locks(&self, instance_id: &str) { - let _ = self.instance_content_locks.remove(instance_id); - let _ = self.instance_screenshot_locks.remove(instance_id); - let _ = self.shared_instance_locks.remove(instance_id); - } - pub async fn init(app_identifier: String) -> crate::Result<()> { let _startup = STATE_STARTUP_LOCK.lock().await; let result = LAUNCHER_STATE @@ -291,7 +241,7 @@ impl State { } let res = tokio::try_join!( - state.discord_rpc.clear_to_default( + state.presence.discord_rpc.clear_to_default( true, &state.pool, &state.process_manager, @@ -309,6 +259,7 @@ impl State { } let _ = state + .presence .friends_socket .connect( &state.pool, @@ -401,15 +352,13 @@ impl State { ) .await?; - let discord_rpc = DiscordGuard::init()?; + let presence = Presence::init()?; tracing::info!("Initializing file watcher"); let file_watcher = instances::watcher::init_watcher().await?; let process_manager = ProcessManager::new(); - let friends_socket = FriendsSocket::new(); - Ok(Arc::new(Self { startup_complete: AtomicBool::new(false), directories, @@ -417,18 +366,12 @@ impl State { fetch_semaphore, io_semaphore, api_semaphore, - install_job_semaphore: Semaphore::new(MAX_CONCURRENT_INSTALL_JOBS), - install_db_semaphore: Semaphore::new(1), - instance_content_locks: DashMap::new(), - instance_screenshot_locks: DashMap::new(), - shared_instance_locks: DashMap::new(), - synced_options_lock: Mutex::new(()), - game_locale_indexer: - crate::api::instance::GameLocaleIndexer::default(), - pack_sync_worker: crate::api::instance::PackSyncWorker::default(), - discord_rpc, + installs: Installs::new(), + instance_locks: InstanceLocks::default(), + synced_options: SyncedOptions::default(), + presence, process_manager, - friends_socket, + restart_after_pending_update: AtomicBool::new(false), pool, file_watcher, // app_identifier, diff --git a/packages/app-lib/src/state/presence.rs b/packages/app-lib/src/state/presence.rs new file mode 100644 index 0000000000..0d89483f73 --- /dev/null +++ b/packages/app-lib/src/state/presence.rs @@ -0,0 +1,16 @@ +use super::{DiscordGuard, FriendsSocket}; + +/// Long-running social integrations. +pub struct Presence { + pub discord_rpc: DiscordGuard, + pub friends_socket: FriendsSocket, +} + +impl Presence { + pub(crate) fn init() -> crate::Result { + Ok(Self { + discord_rpc: DiscordGuard::init()?, + friends_socket: FriendsSocket::new(), + }) + } +} diff --git a/packages/app-lib/src/state/process.rs b/packages/app-lib/src/state/process.rs index 459df60677..fc6e8aa8a8 100644 --- a/packages/app-lib/src/state/process.rs +++ b/packages/app-lib/src/state/process.rs @@ -1033,11 +1033,12 @@ impl Process { } let _ = state + .presence .discord_rpc .clear_to_default(true, &state.pool, &state.process_manager) .await; - let _ = state.friends_socket.update_status(None).await; + let _ = state.presence.friends_socket.update_status(None).await; #[cfg(feature = "tauri")] { diff --git a/packages/app-lib/src/state/synced_options.rs b/packages/app-lib/src/state/synced_options.rs new file mode 100644 index 0000000000..c8aa48cf43 --- /dev/null +++ b/packages/app-lib/src/state/synced_options.rs @@ -0,0 +1,17 @@ +use crate::api::instance::{GameLocaleIndexer, PackSyncWorker}; +use tokio::sync::{Mutex, MutexGuard}; + +/// Coordinates options syncing across instances. +#[derive(Default)] +pub(crate) struct SyncedOptions { + /// Serializes canonical synced-option mutations and checkpoint updates. + lock: Mutex<()>, + pub(crate) locales: GameLocaleIndexer, + pub(crate) packs: PackSyncWorker, +} + +impl SyncedOptions { + pub(crate) async fn lock(&self) -> MutexGuard<'_, ()> { + self.lock.lock().await + } +} diff --git a/packages/app-lib/src/util/fetch.rs b/packages/app-lib/src/util/fetch.rs index d89ffaf381..73a6941327 100644 --- a/packages/app-lib/src/util/fetch.rs +++ b/packages/app-lib/src/util/fetch.rs @@ -2,6 +2,7 @@ use super::io::{self, IOError}; use crate::event::LoadingBarId; use crate::event::emit::emit_loading; +use crate::state::content_store::ContentStore; use crate::util::content_hash::{ContentHasher, temporary_file}; use crate::{ErrorKind, LabrinthError}; use anyhow::{Context, bail}; @@ -427,12 +428,12 @@ impl DownloadedFile { pub(crate) async fn store_file( &self, - state: &crate::State, + content_store: &ContentStore, ) -> crate::Result { match &self.path { DownloadedFilePath::Stored(stored_file) => Ok(stored_file.clone()), DownloadedFilePath::Temporary(_) => { - state.content_store.store_file(self.path()).await + content_store.store_file(self.path()).await } } }