v0.3.7-pre.008

crates/ksp-app-backfill-desk/src/app_state.rs

@@ -1,10 +1,11 @@
 // file: crates/ksp-app-backfill-desk/src/app_state.rs
-// version: 5
+// version: 6
 
 //! Shared backend state owned by the Backfill Desk Tauri application.
 
 /// Shared Backfill Desk application state managed by Tauri.
 pub(crate) struct AppState {
+    backfill_runs: crate::BackfillRunState,
     config_management: ksp_config_lib::ConfigManagement,
     logging_runtime: std::sync::Mutex<LoggingRuntimeState>,
     shutdown_started: std::sync::atomic::AtomicBool,
@@ -76,6 +77,7 @@ impl crate::AppState {
             "resolved Backfill Desk splash timings"
         );
         return std::result::Result::Ok(Self {
+            backfill_runs: crate::BackfillRunState::new(),
             config_management,
             logging_runtime: std::sync::Mutex::new(LoggingRuntimeState {
                 guard: logging_startup.guard,
@@ -121,7 +123,7 @@ impl crate::AppState {
             application_version: env!("CARGO_PKG_VERSION").to_owned(),
             config_document_count: document_count,
             fallback_logging_active: runtime.fallback_active,
-            shell_phase: "pre.007-request-mapping".to_owned(),
+            shell_phase: "pre.008-runtime-start".to_owned(),
             startup_diagnostic: runtime.startup_diagnostic.clone(),
         });
     }
@@ -194,6 +196,103 @@ impl crate::AppState {
         return std::result::Result::Ok(preview);
     }
 
+    /// Prepares one concrete Backfill runtime, installs its control handle atomically and lends execution resources before spawn.
+    pub(crate) fn prepare_backfill_start(&self, request: crate::BackfillStartRequestDto) -> ksp_core_lib::Result<crate::BackfillRunLaunch> {
+        if self.shutdown_started.load(std::sync::atomic::Ordering::Acquire) {
+            return std::result::Result::Err(ksp_core_lib::Error::new(
+                crate::ERROR_CODE_BACKFILL_COMPOSITION_NOT_READY,
+                "Backfill Desk cannot start a Backfill run while application shutdown is in progress",
+            ));
+        }
+        let options = self.backfill_options();
+        let options = match options {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let job_id = self.backfill_runs.next_job_id();
+        let job_id = match job_id {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let mapped = crate::map_backfill_request(request, &options, job_id.clone());
+        let mapped = match mapped {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let runtime = ksp_job_backfill_lib::BackfillJobRuntime::new(mapped);
+        let runtime = match runtime {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let handle = runtime.handle();
+        let installed = self.backfill_runs.install(job_id.clone(), handle);
+        if let std::result::Result::Err(error) = installed {
+            return std::result::Result::Err(error);
+        }
+        let transport = self.transport_runtime.as_ref();
+        let transport = match transport {
+            std::option::Option::Some(value) => value.pool(),
+            std::option::Option::None => {
+                let rollback = self.backfill_runs.rollback(&job_id);
+                if let std::result::Result::Err(error) = rollback {
+                    return std::result::Result::Err(error);
+                }
+                return std::result::Result::Err(ksp_core_lib::Error::new(
+                    crate::ERROR_CODE_BACKFILL_COMPOSITION_NOT_READY,
+                    "Backfill Desk cannot start a Backfill run without a ready HTTP Transport pool",
+                ));
+            },
+        };
+        let store = self.store_startup.take_for_run();
+        let store = match store {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => {
+                let rollback = self.backfill_runs.rollback(&job_id);
+                if let std::result::Result::Err(rollback_error) = rollback {
+                    return std::result::Result::Err(rollback_error);
+                }
+                return std::result::Result::Err(error);
+            },
+        };
+        ksp_logging_lib::debug!(
+            target: crate::TRACING_TARGET,
+            domain = crate::TRACING_DOMAIN_RUN,
+            job_id = job_id.as_str(),
+            "admitted Backfill Desk run and installed control handle before async spawn"
+        );
+        return std::result::Result::Ok(crate::BackfillRunLaunch { job_id, runtime, store, transport });
+    }
+
+    /// Executes one already-admitted Backfill launch and restores application resources after its terminal result.
+    pub(crate) async fn execute_backfill_run(&self, launch: crate::BackfillRunLaunch) -> ksp_core_lib::Result<()> {
+        let job_id = launch.job_id.clone();
+        let result = launch.runtime.run(&launch.transport, &launch.store).await;
+        let restore = self.store_startup.restore_after_run(launch.store);
+        let finished = self.backfill_runs.finish(&job_id);
+        let cancellation_requested = match finished {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        if let std::result::Result::Err(error) = restore {
+            return std::result::Result::Err(error);
+        }
+        return match result {
+            std::result::Result::Ok(snapshot) => {
+                ksp_logging_lib::debug!(
+                    target: crate::TRACING_TARGET,
+                    domain = crate::TRACING_DOMAIN_RUN,
+                    cancellation_requested = cancellation_requested,
+                    contiguous_completed = snapshot.contiguous_completed(),
+                    job_id = job_id.as_str(),
+                    phase = snapshot.phase().code(),
+                    "Backfill Desk run reached a terminal snapshot and released the single-run slot"
+                );
+                std::result::Result::Ok(())
+            },
+            std::result::Result::Err(error) => std::result::Result::Err(error),
+        };
+    }
+
     /// Marks graceful application shutdown as started and reports whether this caller won the one-shot transition.
     pub(crate) fn begin_shutdown(&self) -> bool {
         return self.shutdown_started.compare_exchange(false, true, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire).is_ok();