fix: sqlite lock starvation & excessive row rewrites

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');

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,24 +173,24 @@ 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 {
                 false
             };
 
         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

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<u32>,
     pub target: InstallTarget,
     pub cleanup: InstallCleanup,
     pub progress: InstallProgressState,
@@ -32,15 +35,20 @@ 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();
         let phase = InstallPhaseId::PreparingInstance;
 
         Self {
             schema_version: 1,
-            request,
+            content_count: match request {
+                InstallRequest::BulkUpdateContent { updates, .. } => {
+                    Some(updates.len() as u32)
+                }
+                _ => None,
+            },
             target,
             cleanup,
             progress: InstallProgressState {

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<InstallJobDisplay> {
-    match &state.request {
+fn display_from_request(
+    request: &InstallRequest,
+    state: &InstallJobState,
+) -> Option<InstallJobDisplay> {
+    match request {
         InstallRequest::CreateInstance { name, icon_path, .. } => {
             Some(InstallJobDisplay {
                 title: name.clone(),