diff --git a/packages/app-lib/.sqlx/query-207f7dd38b90a8f4e816eb589edfdbaf0bb8d1e05c816defe9b5f9ab50f64e16.json b/packages/app-lib/.sqlx/query-207f7dd38b90a8f4e816eb589edfdbaf0bb8d1e05c816defe9b5f9ab50f64e16.json new file mode 100644 index 00000000000..c4dd46b8afb --- /dev/null +++ b/packages/app-lib/.sqlx/query-207f7dd38b90a8f4e816eb589edfdbaf0bb8d1e05c816defe9b5f9ab50f64e16.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "\n\t\tSELECT json(request) AS \"request!: String\"\n\t\tFROM install_job_requests\n\t\tWHERE job_id = ?\n\t\t", + "describe": { + "columns": [ + { + "name": "request!: String", + "ordinal": 0, + "type_info": "Null" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + null + ] + }, + "hash": "207f7dd38b90a8f4e816eb589edfdbaf0bb8d1e05c816defe9b5f9ab50f64e16" +} diff --git a/packages/app-lib/.sqlx/query-a8dee31ead799ef55e45aebbbf537dc2520f33ddec5bba5d844e7de2c8ee020e.json b/packages/app-lib/.sqlx/query-a8dee31ead799ef55e45aebbbf537dc2520f33ddec5bba5d844e7de2c8ee020e.json new file mode 100644 index 00000000000..12db0b4cc8d --- /dev/null +++ b/packages/app-lib/.sqlx/query-a8dee31ead799ef55e45aebbbf537dc2520f33ddec5bba5d844e7de2c8ee020e.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n\t\tDELETE FROM install_job_requests\n\t\tWHERE job_id = ?\n\t\t", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "a8dee31ead799ef55e45aebbbf537dc2520f33ddec5bba5d844e7de2c8ee020e" +} diff --git a/packages/app-lib/.sqlx/query-fd02c7fe5b0fbf20a35177500040da32cf608e10628b7491450eff31c6c39eba.json b/packages/app-lib/.sqlx/query-fd02c7fe5b0fbf20a35177500040da32cf608e10628b7491450eff31c6c39eba.json new file mode 100644 index 00000000000..cb38c8a102e --- /dev/null +++ b/packages/app-lib/.sqlx/query-fd02c7fe5b0fbf20a35177500040da32cf608e10628b7491450eff31c6c39eba.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n\t\tINSERT INTO install_job_requests (job_id, request)\n\t\tVALUES (?, ?)\n\t\t", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "fd02c7fe5b0fbf20a35177500040da32cf608e10628b7491450eff31c6c39eba" +} diff --git a/packages/app-lib/migrations/20261022120000_install-job-requests.sql b/packages/app-lib/migrations/20261022120000_install-job-requests.sql new file mode 100644 index 00000000000..2dd09cdc828 --- /dev/null +++ b/packages/app-lib/migrations/20261022120000_install-job-requests.sql @@ -0,0 +1,21 @@ +CREATE TABLE install_job_requests ( + job_id TEXT PRIMARY KEY NOT NULL REFERENCES install_jobs(id) ON DELETE CASCADE, + request JSONB NOT NULL +); + +INSERT INTO install_job_requests (job_id, request) +SELECT id, json_extract(state, '$.request') +FROM install_jobs +WHERE status NOT IN ('succeeded', 'canceled') + AND json_extract(state, '$.request') IS NOT NULL; + +UPDATE install_jobs +SET state = json_set( + state, + '$.content_count', + json_array_length(state, '$.request.updates') +) +WHERE json_extract(state, '$.request.type') = 'bulk_update_content'; + +UPDATE install_jobs +SET state = json_remove(state, '$.request'); diff --git a/packages/app-lib/src/install/events.rs b/packages/app-lib/src/install/events.rs index d49abbd9107..dbc61a46c9e 100644 --- a/packages/app-lib/src/install/events.rs +++ b/packages/app-lib/src/install/events.rs @@ -9,8 +9,8 @@ use std::time::{Duration, Instant}; use tokio::sync::Mutex; use uuid::Uuid; -const PROGRESS_PERSIST_INTERVAL: Duration = Duration::from_millis(750); -const CONTENT_PROGRESS_PERSIST_STEPS: u64 = 25; +const PROGRESS_EMIT_INTERVAL: Duration = Duration::from_millis(750); +const CONTENT_PROGRESS_EMIT_STEPS: u64 = 25; #[derive(Clone, Debug)] pub struct InstallProgressReporter { @@ -22,8 +22,8 @@ pub struct InstallProgressReporter { #[derive(Debug)] struct InstallProgressReporterState { job: InstallJobState, - last_persisted_at: Instant, - last_persisted_progress: Option<(InstallPhaseId, u64)>, + last_emitted_at: Instant, + last_emitted_progress: Option<(InstallPhaseId, u64)>, } impl InstallProgressReporter { @@ -33,8 +33,8 @@ impl InstallProgressReporter { pool, state: Arc::new(Mutex::new(InstallProgressReporterState { job: state, - last_persisted_at: Instant::now(), - last_persisted_progress: None, + last_emitted_at: Instant::now(), + last_emitted_progress: None, })), } } @@ -83,7 +83,7 @@ impl InstallProgressReporter { let record = store::update_state(self.job_id, &state.job, &self.pool).await?; - state.mark_persisted(); + state.mark_emitted(); emit_install_job(&record.snapshot()).await } @@ -92,7 +92,7 @@ impl InstallProgressReporter { let record = store::update_state(self.job_id, &state.job, &self.pool).await?; - state.mark_persisted(); + state.mark_emitted(); let snapshot = record.snapshot(); emit_install_job(&snapshot).await?; Ok(snapshot) @@ -135,28 +135,34 @@ impl InstallProgressReporter { InstallPhaseDetails::Empty ) && !matches!(&details, InstallPhaseDetails::Empty); + let persist = phase_started || !events.is_empty(); + state.job.set_progress(phase, progress, details); for event in events { state.job.record_event(event); } - if !state.should_persist(phase_started) { - return Ok(()); + if !persist { + if !state.should_emit() { + return Ok(()); + } + if let Some(snapshot) = + store::update_live_progress(self.job_id, &state.job) + { + state.mark_emitted(); + return emit_install_job(&snapshot).await; + } } let record = store::update_state(self.job_id, &state.job, &self.pool).await?; - state.mark_persisted(); + state.mark_emitted(); emit_install_job(&record.snapshot()).await } } impl InstallProgressReporterState { - fn should_persist(&self, phase_started: bool) -> bool { - if phase_started { - return true; - } - + fn should_emit(&self) -> bool { let Some(progress) = &self.job.progress.progress else { return true; }; @@ -167,11 +173,11 @@ impl InstallProgressReporterState { let progressed_enough = if self.job.progress.phase == InstallPhaseId::DownloadingContent { - self.last_persisted_progress + self.last_emitted_progress .map(|(phase, current)| { phase != self.job.progress.phase || progress.current.saturating_sub(current) - >= CONTENT_PROGRESS_PERSIST_STEPS + >= CONTENT_PROGRESS_EMIT_STEPS }) .unwrap_or(true) } else { @@ -179,12 +185,12 @@ impl InstallProgressReporterState { }; progressed_enough - || self.last_persisted_at.elapsed() >= PROGRESS_PERSIST_INTERVAL + || self.last_emitted_at.elapsed() >= PROGRESS_EMIT_INTERVAL } - fn mark_persisted(&mut self) { - self.last_persisted_at = Instant::now(); - self.last_persisted_progress = self + fn mark_emitted(&mut self) { + self.last_emitted_at = Instant::now(); + self.last_emitted_progress = self .job .progress .progress diff --git a/packages/app-lib/src/install/model.rs b/packages/app-lib/src/install/model.rs index 413d044fd4b..15a0942c884 100644 --- a/packages/app-lib/src/install/model.rs +++ b/packages/app-lib/src/install/model.rs @@ -11,10 +11,13 @@ use std::path::PathBuf; pub type InstallModpackPreview = CreatePackInstance; +/// Everything about a job except its request, which is stored separately +/// because it can be several megabytes and is only needed to run the job. #[derive(Serialize, Deserialize, Clone, Debug)] pub struct InstallJobState { pub schema_version: u32, - pub request: InstallRequest, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content_count: Option, pub target: InstallTarget, pub cleanup: InstallCleanup, pub progress: InstallProgressState, @@ -32,7 +35,7 @@ pub struct InstallJobState { } impl InstallJobState { - pub fn new(request: InstallRequest) -> Self { + pub fn new(request: &InstallRequest) -> Self { let target = request.target(); let cleanup = request.cleanup(); let kind = request.kind(); @@ -40,7 +43,12 @@ impl InstallJobState { Self { schema_version: 1, - request, + content_count: match request { + InstallRequest::BulkUpdateContent { updates, .. } => { + Some(updates.len() as u32) + } + _ => None, + }, target, cleanup, progress: InstallProgressState { diff --git a/packages/app-lib/src/install/recovery.rs b/packages/app-lib/src/install/recovery.rs index d1137a57f82..55f2af795ad 100644 --- a/packages/app-lib/src/install/recovery.rs +++ b/packages/app-lib/src/install/recovery.rs @@ -206,14 +206,17 @@ fn instance_update_backup_dir( async fn recover_unrecorded_instance_update_backup( job: &mut store::InstallJobRecord, + request: Option<&InstallRequest>, state: &State, ) -> crate::Result<()> { if job.state.paths.staging_dir.is_some() || !matches!( - &job.state.request, - InstallRequest::BulkUpdateContent { .. } - | InstallRequest::UpdateSharedInstance { .. } - | InstallRequest::InstallPackToExistingInstance { .. } + request, + Some( + InstallRequest::BulkUpdateContent { .. } + | InstallRequest::UpdateSharedInstance { .. } + | InstallRequest::InstallPackToExistingInstance { .. } + ) ) { return Ok(()); @@ -659,9 +662,17 @@ async fn recover_interrupted_job_inner( mut job: store::InstallJobRecord, state: &State, ) -> crate::Result<()> { - recover_unrecorded_instance_update_backup(&mut job, state).await?; + let request = store::get_request(job.id, &state.pool).await?; + recover_unrecorded_instance_update_backup( + &mut job, + request.as_ref(), + state, + ) + .await?; if job.state.display.is_none() { - job.state.display = display_from_request(&job.state); + job.state.display = request + .as_ref() + .and_then(|request| display_from_request(request, &job.state)); } if let Some(instance_id) = target_instance_id(&job.state.target) @@ -770,8 +781,11 @@ fn clear_deleted_new_instance_id(job_state: &mut InstallJobState) { } } -fn display_from_request(state: &InstallJobState) -> Option { - match &state.request { +fn display_from_request( + request: &InstallRequest, + state: &InstallJobState, +) -> Option { + match request { InstallRequest::CreateInstance { name, icon_path, .. } => { Some(InstallJobDisplay { title: name.clone(), diff --git a/packages/app-lib/src/install/runner.rs b/packages/app-lib/src/install/runner.rs index 68ef9eae75d..45c8ab35361 100644 --- a/packages/app-lib/src/install/runner.rs +++ b/packages/app-lib/src/install/runner.rs @@ -253,6 +253,7 @@ async fn retry_job_inner(job_id: Uuid) -> crate::Result { .to_string(), )); } + let request = store::get_required_request(job_id, &state.pool).await?; if let Some(instance_id) = current_instance_id(&job.state) { store::ensure_no_pending_recovery( @@ -276,9 +277,9 @@ async fn retry_job_inner(job_id: Uuid) -> crate::Result { } drop(cleanup_target_guard); - let mut target_guard = reserve_target(&job.state.request.target())?; - job.state.target = job.state.request.target(); - job.state.cleanup = job.state.request.cleanup(); + let mut target_guard = reserve_target(&request.target())?; + job.state.target = request.target(); + job.state.cleanup = request.cleanup(); job.state.rollback = None; job.state.paths.staging_dir = None; job.state.error = None; @@ -288,7 +289,7 @@ async fn retry_job_inner(job_id: Uuid) -> crate::Result { job.state.progress.progress = None; job.state.progress.details = InstallPhaseDetails::Empty; job.state.record_event(InstallJobEventKind::JobQueued { - kind: job.state.request.kind(), + kind: request.kind(), }); let record = match store::update_status( @@ -313,7 +314,9 @@ async fn retry_job_inner(job_id: Uuid) -> crate::Result { }; emit_install_job(&record.snapshot()).await?; - if let Err(error) = prepare_initial_instance(&mut job.state, &state).await { + if let Err(error) = + prepare_initial_instance(&request, &mut job.state, &state).await + { let error_view = install_error_view( job.state.progress.phase, &error, @@ -494,14 +497,21 @@ async fn start_inner( .await?; } let id = Uuid::new_v4(); - let mut job_state = InstallJobState::new(request); - set_initial_display(&mut job_state); - let record = - store::insert(id, &job_state, InstallJobStatus::Queued, &state.pool) - .await?; + let mut job_state = InstallJobState::new(&request); + set_initial_display(&request, &mut job_state); + let record = store::insert( + id, + &request, + &job_state, + InstallJobStatus::Queued, + &state.pool, + ) + .await?; emit_install_job(&record.snapshot()).await?; - if let Err(error) = prepare_initial_instance(&mut job_state, &state).await { + if let Err(error) = + prepare_initial_instance(&request, &mut job_state, &state).await + { let error_view = install_error_view( job_state.progress.phase, &error, @@ -559,17 +569,19 @@ async fn start_inner( } fn prepare_initial_instance<'a>( + request: &'a InstallRequest, job_state: &'a mut InstallJobState, state: &'a State, ) -> impl Future> + Send + 'a { - Box::pin(prepare_initial_instance_inner(job_state, state)) + Box::pin(prepare_initial_instance_inner(request, job_state, state)) } async fn prepare_initial_instance_inner( + request: &InstallRequest, job_state: &mut InstallJobState, state: &State, ) -> crate::Result<()> { - match job_state.request.clone() { + match request.clone() { InstallRequest::CreateInstance { name, game_version, @@ -743,8 +755,13 @@ async fn prepare_initial_instance_inner( } | InstallRequest::BulkUpdateContent { instance_id, .. } | InstallRequest::UpdateSharedInstance { instance_id, .. } => { - prepare_existing_rollback(job_state, &state.pool, &instance_id) - .await?; + prepare_existing_rollback( + request, + job_state, + &state.pool, + &instance_id, + ) + .await?; } } @@ -807,6 +824,7 @@ async fn run_job_inner( return Ok(()); } + let request = store::get_required_request(job_id, &state.pool).await?; let mut job_state = job.state.clone(); job_state.record_event(InstallJobEventKind::JobStarted); let Some(record) = store::update_status_if( @@ -827,7 +845,7 @@ async fn run_job_inner( super::control::CURRENT_INSTALL .scope( control.clone(), - run_request(job_id, &mut job_state, state), + run_request(job_id, &request, &mut job_state, state), ) .await } @@ -884,10 +902,7 @@ async fn run_job_inner( "Failed to reconcile synced options after installing {instance_id}: {error}" ); } - if matches!( - job_state.request, - InstallRequest::BulkUpdateContent { .. } - ) { + if matches!(request, InstallRequest::BulkUpdateContent { .. }) { Box::pin(crate::api::instance::synced_packs::reconcile_after_content_change( &instance_id, )).await; @@ -1035,18 +1050,20 @@ async fn terminalize_failed_job_inner( fn run_request<'a>( job_id: Uuid, + request: &'a InstallRequest, job_state: &'a mut InstallJobState, state: &'a State, ) -> impl Future>> + Send + 'a { - Box::pin(run_request_inner(job_id, job_state, state)) + Box::pin(run_request_inner(job_id, request, job_state, state)) } async fn run_request_inner( job_id: Uuid, + request: &InstallRequest, job_state: &mut InstallJobState, state: &State, ) -> crate::Result> { - match job_state.request.clone() { + match request.clone() { InstallRequest::CreateInstance { name, game_version, @@ -1239,8 +1256,13 @@ async fn run_request_inner( Ok(Some(instance_id)) } InstallRequest::InstallExistingInstance { instance_id, force } => { - prepare_existing_rollback(job_state, &state.pool, &instance_id) - .await?; + prepare_existing_rollback( + request, + job_state, + &state.pool, + &instance_id, + ) + .await?; lock_instance(&instance_id, state).await?; update_progress( job_id, @@ -1275,10 +1297,15 @@ async fn run_request_inner( location, post_install_edit, } => { - prepare_existing_rollback(job_state, &state.pool, &instance_id) - .await?; + prepare_existing_rollback( + request, + job_state, + &state.pool, + &instance_id, + ) + .await?; lock_instance(&instance_id, state).await?; - prepare_update_backup(job_id, job_state, state).await?; + prepare_update_backup(job_id, request, job_state, state).await?; crate::api::instance::prepare_instance_update(&instance_id).await?; let disabled_project_ids = remove_existing_pack_content( job_id, @@ -1326,7 +1353,7 @@ async fn run_request_inner( state, )) .await?; - prepare_update_backup(job_id, job_state, state).await?; + prepare_update_backup(job_id, request, job_state, state).await?; crate::state::instances::commands::apply_bulk_update( &instance_id, plan, @@ -1341,10 +1368,15 @@ async fn run_request_inner( Ok(Some(instance_id)) } InstallRequest::UpdateSharedInstance { instance_id, data } => { - prepare_existing_rollback(job_state, &state.pool, &instance_id) - .await?; + prepare_existing_rollback( + request, + job_state, + &state.pool, + &instance_id, + ) + .await?; lock_instance(&instance_id, state).await?; - prepare_update_backup(job_id, job_state, state).await?; + prepare_update_backup(job_id, request, job_state, state).await?; let disabled_project_ids = disabled_project_ids(&instance_id, state).await?; apply_shared_instance_update( @@ -1648,6 +1680,7 @@ async fn install_pack_inner( } async fn prepare_existing_rollback( + request: &InstallRequest, job_state: &mut InstallJobState, pool: &SqlitePool, instance_id: &str, @@ -1664,7 +1697,7 @@ async fn prepare_existing_rollback( "content in quarantined instances cannot be changed".to_string(), )); } - if matches!(job_state.request, InstallRequest::BulkUpdateContent { .. }) + if matches!(request, InstallRequest::BulkUpdateContent { .. }) && instance.instance.install_stage != InstanceInstallStage::Installed { bail!(ErrorKind::InputError( @@ -1690,11 +1723,13 @@ async fn prepare_existing_rollback( async fn prepare_update_backup( job_id: Uuid, + request: &InstallRequest, job_state: &mut InstallJobState, state: &State, ) -> crate::Result<()> { super::control::checkpoint(job_id).await?; - let scoped_paths = update_backup_scope(job_id, job_state, state).await?; + let scoped_paths = + update_backup_scope(job_id, request, job_state, state).await?; let rollback = job_state .rollback .as_ref() @@ -1719,13 +1754,14 @@ async fn prepare_update_backup( /// hash, so installing the pack afterwards reuses the download. async fn update_backup_scope( job_id: Uuid, + request: &InstallRequest, job_state: &InstallJobState, state: &State, ) -> crate::Result>> { let mut paths = ProjectType::iterator() .map(|project_type| project_type.get_folder().to_string()) .collect::>(); - let (instance_id, packs) = match &job_state.request { + let (instance_id, packs) = match request { InstallRequest::BulkUpdateContent { .. } => return Ok(Some(paths)), InstallRequest::InstallPackToExistingInstance { instance_id, @@ -1914,6 +1950,8 @@ pub(super) async fn update_content_progress_with_bytes( bytes: Option, ) -> crate::Result<()> { super::control::checkpoint(job_id).await?; + let phase_started = + job_state.progress.phase != InstallPhaseId::DownloadingContent; job_state.progress.phase = InstallPhaseId::DownloadingContent; job_state.progress.progress = Some(InstallProgress { current, @@ -1921,6 +1959,11 @@ pub(super) async fn update_content_progress_with_bytes( secondary: bytes, }); job_state.progress.details = InstallPhaseDetails::Empty; + if !phase_started + && let Some(snapshot) = store::update_live_progress(job_id, job_state) + { + return emit_install_job(&snapshot).await; + } let record = store::update_state(job_id, job_state, pool).await?; emit_install_job(&record.snapshot()).await?; Ok(()) @@ -1965,8 +2008,11 @@ fn set_display( job_state.display = Some(InstallJobDisplay { title, icon }); } -fn set_initial_display(job_state: &mut InstallJobState) { - let display = match &job_state.request { +fn set_initial_display( + request: &InstallRequest, + job_state: &mut InstallJobState, +) { + let display = match request { InstallRequest::CreateModpackInstance { location, .. } => { match location { CreatePackLocation::FromVersionId { diff --git a/packages/app-lib/src/install/store.rs b/packages/app-lib/src/install/store.rs index aae10f00907..58351819274 100644 --- a/packages/app-lib/src/install/store.rs +++ b/packages/app-lib/src/install/store.rs @@ -1,12 +1,13 @@ use super::model::{ InstallJobKind, InstallJobSnapshot, InstallJobState, InstallJobStatus, + InstallProgressState, InstallRequest, }; use crate::ErrorKind; use anyhow::{Context, bail}; use chrono::{DateTime, TimeZone, Utc}; use sqlx::SqlitePool; use std::collections::HashMap; -use std::sync::{Arc, LazyLock, Mutex, Weak}; +use std::sync::{Arc, LazyLock, Mutex, MutexGuard, Weak}; use tokio::sync::Notify; use uuid::Uuid; @@ -71,29 +72,19 @@ impl InstallJobRecord { && self.state.rollback_error.is_some() } + /// Active jobs report their live progress, which can be newer than the stored one. pub fn snapshot(&self) -> InstallJobSnapshot { - let (paused, canceling, controllable) = - super::control::snapshot(self.id); - let active = matches!( - self.status, - InstallJobStatus::Queued | InstallJobStatus::Running - ); - InstallJobSnapshot { - content_count: match &self.state.request { - super::model::InstallRequest::BulkUpdateContent { - updates, - .. - } => Some(updates.len() as u32), - _ => None, - }, + let active = is_active(self.status); + let mut snapshot = InstallJobSnapshot { + content_count: self.state.content_count, job_id: self.id.to_string(), instance_id: self.instance_id.clone(), kind: self.kind, status: self.status, - paused: active && paused, - canceling: active && canceling, - can_pause: active && controllable, - can_cancel: active && controllable, + paused: false, + canceling: false, + can_pause: false, + can_cancel: false, target: self.state.target.clone(), phase: self.state.progress.phase, progress: self.state.progress.progress.clone(), @@ -104,19 +95,104 @@ impl InstallJobRecord { created: self.created.to_rfc3339(), modified: self.modified.to_rfc3339(), finished: self.finished.map(|finished| finished.to_rfc3339()), + }; + set_controls(&mut snapshot, self.id, active); + if active && let Some(job) = live_jobs().get_mut(&self.id) { + job.apply_progress(&mut snapshot); + job.snapshot = Some(snapshot.clone()); } + snapshot } } +/// Progress of jobs running in this process. It changes far more often than the +/// rest of a job, so it is kept here instead of rewriting the stored job on +/// every update. +static LIVE_JOBS: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +struct LiveJob { + progress: InstallProgressState, + snapshot: Option, +} + +impl LiveJob { + fn apply_progress(&self, snapshot: &mut InstallJobSnapshot) { + snapshot.phase = self.progress.phase; + snapshot.progress = self.progress.progress.clone(); + snapshot.details = self.progress.details.clone(); + } +} + +fn live_jobs() -> MutexGuard<'static, HashMap> { + LIVE_JOBS.lock().unwrap_or_else(|error| error.into_inner()) +} + +fn is_active(status: InstallJobStatus) -> bool { + matches!(status, InstallJobStatus::Queued | InstallJobStatus::Running) +} + +fn set_controls(snapshot: &mut InstallJobSnapshot, id: Uuid, active: bool) { + let (paused, canceling, controllable) = super::control::snapshot(id); + snapshot.paused = active && paused; + snapshot.canceling = active && canceling; + snapshot.can_pause = active && controllable; + snapshot.can_cancel = active && controllable; +} + +/// Keeps the live progress of a job in step with what was just stored for it. +fn track_live_job(record: &InstallJobRecord) { + let mut jobs = live_jobs(); + if !is_active(record.status) { + jobs.remove(&record.id); + return; + } + let progress = record.state.progress.clone(); + jobs.entry(record.id) + .and_modify(|job| job.progress = progress.clone()) + .or_insert(LiveJob { + progress, + snapshot: None, + }); +} + +/// Updates the progress of a running job without storing it. Returns `None` +/// when the job has no snapshot to update yet, so the caller should store it. +pub(super) fn update_live_progress( + id: Uuid, + state: &InstallJobState, +) -> Option { + let mut jobs = live_jobs(); + let job = jobs.get_mut(&id)?; + let mut snapshot = job.snapshot.clone()?; + job.progress = state.progress.clone(); + job.apply_progress(&mut snapshot); + set_controls(&mut snapshot, id, true); + job.snapshot = Some(snapshot.clone()); + Some(snapshot) +} + +/// Reads back a job that was just written. +async fn get_written( + id: Uuid, + pool: &SqlitePool, +) -> crate::Result { + let record = get_required(id, pool).await?; + track_live_job(&record); + Ok(record) +} + pub async fn insert( id: Uuid, + request: &InstallRequest, state: &InstallJobState, status: InstallJobStatus, pool: &SqlitePool, ) -> crate::Result { let now = Utc::now(); - let kind = state.request.kind(); + let kind = request.kind(); let json = serde_json::to_string(state)?; + let request_json = serde_json::to_string(request)?; let status_value = status.as_str(); let kind_value = kind.as_str(); let instance_id = instance_id(state); @@ -124,6 +200,7 @@ pub async fn insert( let created = now.timestamp(); let modified = created; + let mut transaction = pool.begin_with("BEGIN IMMEDIATE").await?; sqlx::query!( " INSERT INTO install_jobs ( @@ -139,14 +216,86 @@ pub async fn insert( created, modified, ) - .execute(pool) + .execute(&mut *transaction) + .await?; + sqlx::query!( + " + INSERT INTO install_job_requests (job_id, request) + VALUES (?, ?) + ", + id_value, + request_json, + ) + .execute(&mut *transaction) .await?; + transaction.commit().await?; get(id, pool) .await? .context(format!("install job {id} was not inserted")) } +/// The request a job was started with. Jobs that succeeded or were canceled +/// no longer have one, since they cannot be retried. +pub async fn get_request( + id: Uuid, + pool: &SqlitePool, +) -> crate::Result> { + let id = id.to_string(); + let request = sqlx::query_scalar!( + " + SELECT json(request) AS \"request!: String\" + FROM install_job_requests + WHERE job_id = ? + ", + id, + ) + .fetch_optional(pool) + .await?; + + Ok(request + .map(|request| serde_json::from_str(&request)) + .transpose()?) +} + +pub async fn get_required_request( + id: Uuid, + pool: &SqlitePool, +) -> crate::Result { + get_request(id, pool).await?.ok_or( + ErrorKind::InputError(format!( + "install job {id} no longer has its request" + )) + .into(), + ) +} + +/// Jobs that succeeded or were canceled cannot be retried, so their request is +/// no longer needed. +async fn release_request( + id: Uuid, + status: InstallJobStatus, + pool: &SqlitePool, +) -> crate::Result<()> { + if !matches!( + status, + InstallJobStatus::Succeeded | InstallJobStatus::Canceled + ) { + return Ok(()); + } + let id = id.to_string(); + sqlx::query!( + " + DELETE FROM install_job_requests + WHERE job_id = ? + ", + id, + ) + .execute(pool) + .await?; + Ok(()) +} + pub async fn get( id: Uuid, pool: &SqlitePool, @@ -320,7 +469,7 @@ pub async fn update_state( ))); } - get_required(id, pool).await + get_written(id, pool).await } pub async fn update_status( @@ -353,11 +502,12 @@ pub async fn update_status( .execute(pool) .await?; + release_request(id, status, pool).await?; if status.is_finished() { notify_completion(id); } - get_required(id, pool).await + get_written(id, pool).await } pub async fn update_status_if( @@ -397,11 +547,12 @@ pub async fn update_status_if( return Ok(None); } + release_request(id, status, pool).await?; if status.is_finished() { notify_completion(id); } - get_required(id, pool).await.map(Some) + get_written(id, pool).await.map(Some) } pub async fn finish_active( @@ -443,11 +594,12 @@ pub async fn finish_active( return Ok(None); } + release_request(id, status, pool).await?; if status.is_finished() { notify_completion(id); } - get_required(id, pool).await.map(Some) + get_written(id, pool).await.map(Some) } pub async fn complete_success( @@ -510,9 +662,10 @@ pub async fn complete_success( } transaction.commit().await?; + release_request(id, InstallJobStatus::Succeeded, pool).await?; notify_completion(id); crate::api::instance::queue_game_locale_index(); - get_required(id, pool).await.map(Some) + get_written(id, pool).await.map(Some) } pub async fn dismiss(id: Uuid, pool: &SqlitePool) -> crate::Result<()> {