449 lines
22 KiB
Rust
449 lines
22 KiB
Rust
// file: crates/ksp-app-backfill-desk/src/app_state.rs
|
|
// version: 10
|
|
|
|
//! 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,
|
|
splash_settings: crate::SplashSettings,
|
|
splash_sequence_started: std::sync::atomic::AtomicBool,
|
|
store_startup: crate::StoreStartup,
|
|
transport_runtime: std::option::Option<crate::TransportRuntime>,
|
|
transport_startup_diagnostic: std::option::Option<crate::CommandErrorDto>,
|
|
}
|
|
|
|
impl crate::AppState {
|
|
/// Initializes the scaffold Config ownership, Logging runtime and splash settings.
|
|
pub(crate) fn initialize(arguments: &[std::ffi::OsString]) -> ksp_core_lib::Result<Self> {
|
|
let config_management = crate::config_management(arguments);
|
|
let config_management = match config_management {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let runtime_identity = crate::launch_identity();
|
|
let runtime_identity = match runtime_identity {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let logging_startup = crate::initialize_logging(&config_management, &runtime_identity);
|
|
let logging_startup = match logging_startup {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let transport_startup = crate::initialize_transport(&config_management);
|
|
let (transport_runtime, transport_startup_diagnostic) = match transport_startup {
|
|
std::result::Result::Ok(value) => (std::option::Option::Some(value), std::option::Option::None),
|
|
std::result::Result::Err(error) => {
|
|
let diagnostic = crate::CommandErrorDto::from_error(&error);
|
|
ksp_logging_lib::warn!(
|
|
target: crate::TRACING_TARGET,
|
|
domain = crate::TRACING_DOMAIN_TRANSPORT,
|
|
error_domain = diagnostic.domain.as_str(),
|
|
error_code = diagnostic.code.as_str(),
|
|
"Backfill Desk Transport readiness is unavailable; keeping desktop shell available"
|
|
);
|
|
(std::option::Option::None, std::option::Option::Some(diagnostic))
|
|
},
|
|
};
|
|
let store_startup = tauri::async_runtime::block_on(crate::initialize_store(&config_management, transport_runtime.as_ref()));
|
|
let splash_settings = crate::SplashSettings::load();
|
|
let splash_settings = match splash_settings {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
ksp_logging_lib::warn!(
|
|
target: crate::TRACING_TARGET,
|
|
domain = crate::TRACING_DOMAIN_WINDOWS,
|
|
error_domain = error.code().domain(),
|
|
error_code = error.code().code(),
|
|
"managed splash timings are invalid; using transient in-memory defaults"
|
|
);
|
|
crate::SplashSettings::fallback()
|
|
},
|
|
};
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
domain = crate::TRACING_DOMAIN_WINDOWS,
|
|
minimum_ms = splash_settings.minimum_ms(),
|
|
minimum_source = splash_settings.minimum_source(),
|
|
fade_in_ms = splash_settings.fade_in_ms(),
|
|
fade_in_source = splash_settings.fade_in_source(),
|
|
fade_out_ms = splash_settings.fade_out_ms(),
|
|
fade_out_source = splash_settings.fade_out_source(),
|
|
expected_backend_lifecycle_ms = splash_settings.expected_backend_lifecycle_ms(),
|
|
"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,
|
|
active_profile_id: logging_startup.active_profile_id,
|
|
fallback_active: logging_startup.fallback_active,
|
|
startup_diagnostic: logging_startup.startup_diagnostic,
|
|
}),
|
|
shutdown_started: std::sync::atomic::AtomicBool::new(false),
|
|
splash_settings,
|
|
splash_sequence_started: std::sync::atomic::AtomicBool::new(false),
|
|
store_startup,
|
|
transport_runtime,
|
|
transport_startup_diagnostic,
|
|
});
|
|
}
|
|
|
|
/// Builds the safe scaffold status exposed by the Backfill Desk shell.
|
|
pub(crate) fn shell_status(&self) -> ksp_core_lib::Result<crate::ShellStatusDto> {
|
|
let document_count = self.config_management.engine().registry().descriptors().count();
|
|
let document_count = u32::try_from(document_count);
|
|
let document_count = match document_count {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_APP_STATE_INVALID, "Config registry count cannot be represented by Backfill Desk")
|
|
.with_source(error),
|
|
);
|
|
},
|
|
};
|
|
let runtime = self.logging_runtime.lock();
|
|
let runtime = match runtime {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => {
|
|
return std::result::Result::Err(ksp_core_lib::Error::new(
|
|
crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
|
|
"Backfill Desk Logging runtime state lock is poisoned",
|
|
));
|
|
},
|
|
};
|
|
let _keep_guard_alive = &runtime.guard;
|
|
return std::result::Result::Ok(crate::ShellStatusDto {
|
|
active_logging_profile: runtime.active_profile_id.clone(),
|
|
application_version: env!("CARGO_PKG_VERSION").to_owned(),
|
|
config_document_count: document_count,
|
|
fallback_logging_active: runtime.fallback_active,
|
|
shell_phase: "pre.012-frontend-polish".to_owned(),
|
|
startup_diagnostic: runtime.startup_diagnostic.clone(),
|
|
});
|
|
}
|
|
|
|
/// Builds the safe Transport-and-Store readiness subset of Backfill Desk options.
|
|
pub(crate) fn backfill_options(&self) -> ksp_core_lib::Result<crate::BackfillDeskOptionsDto> {
|
|
let runtime = self.transport_runtime.as_ref();
|
|
let options = match runtime {
|
|
std::option::Option::Some(value) => value.options(),
|
|
std::option::Option::None => {
|
|
let limits = crate::backfill_request_limits();
|
|
let limits = match limits {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
std::result::Result::Ok(crate::BackfillDeskOptionsDto {
|
|
commitments: crate::backfill_commitment_codes(),
|
|
composition_ready: false,
|
|
configured_networks: std::vec::Vec::new(),
|
|
http_routes: std::vec::Vec::new(),
|
|
limits,
|
|
network_coherent: false,
|
|
program_id_options: crate::program_id_autocomplete_options(),
|
|
scope_kinds: crate::backfill_scope_kind_codes(),
|
|
store_diagnostic: std::option::Option::None,
|
|
store_network: std::option::Option::None,
|
|
store_ready: false,
|
|
transport_diagnostic: self.transport_startup_diagnostic.clone(),
|
|
transport_ready: false,
|
|
})
|
|
},
|
|
};
|
|
let mut options = match options {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
self.store_startup.apply_to(&mut options);
|
|
return std::result::Result::Ok(options);
|
|
}
|
|
|
|
/// Validates and maps one frontend campaign request without starting a Backfill Job.
|
|
pub(crate) fn validate_backfill_request(&self, request: crate::BackfillStartRequestDto) -> ksp_core_lib::Result<crate::BackfillRequestPreviewDto> {
|
|
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 = ksp_job_api::JobId::new("backfill-desk-validation");
|
|
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);
|
|
let mapped = match mapped {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let preview = crate::project_backfill_request(&mapped);
|
|
let preview = match preview {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
domain = crate::TRACING_DOMAIN_REQUEST,
|
|
commitment = preview.commitment.as_str(),
|
|
http_role = preview.http_role.as_str(),
|
|
scope_kind = preview.scope_kind.as_str(),
|
|
"validated Backfill Desk campaign request without starting a Job"
|
|
);
|
|
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 retained_request = mapped.clone();
|
|
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 snapshots = handle.snapshots();
|
|
let installed = self.backfill_runs.install(job_id.clone(), handle, retained_request);
|
|
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, snapshots, store, transport });
|
|
}
|
|
|
|
/// Prepares one in-session Resume from the retained Rust-only checkpoint using a new backend Job identity.
|
|
pub(crate) fn prepare_backfill_resume(&self) -> 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 resume 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),
|
|
};
|
|
if !options.composition_ready {
|
|
return std::result::Result::Err(ksp_core_lib::Error::new(
|
|
crate::ERROR_CODE_BACKFILL_COMPOSITION_NOT_READY,
|
|
"Backfill Desk cannot resume without ready Transport and Store composition",
|
|
));
|
|
}
|
|
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 = self.backfill_runs.resume_request(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 store_network_matches = options.store_network.as_deref() == std::option::Option::Some(mapped.network().as_str());
|
|
let role_available = options.http_routes.iter().any(|route| return route.role == mapped.role().as_str());
|
|
if !store_network_matches || !role_available {
|
|
return std::result::Result::Err(ksp_core_lib::Error::new(
|
|
crate::ERROR_CODE_BACKFILL_COMPOSITION_NOT_READY,
|
|
"Backfill Desk retained Resume request no longer matches current Transport/Store composition",
|
|
));
|
|
}
|
|
let retained_request = mapped.clone();
|
|
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 snapshots = handle.snapshots();
|
|
let installed = self.backfill_runs.install(job_id.clone(), handle, retained_request);
|
|
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 resume 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 in-session Resume with a reissued Rust-only checkpoint"
|
|
);
|
|
return std::result::Result::Ok(crate::BackfillRunLaunch { job_id, runtime, snapshots, store, transport });
|
|
}
|
|
|
|
/// Returns the current active or retained terminal Backfill status for frontend resynchronization.
|
|
pub(crate) fn backfill_status(&self) -> ksp_core_lib::Result<std::option::Option<crate::BackfillRunStatusDto>> {
|
|
let notification = self.backfill_runs.current_notification();
|
|
let notification = match notification {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return match notification {
|
|
std::option::Option::Some(value) => crate::project_backfill_status(&value).map(std::option::Option::Some),
|
|
std::option::Option::None => std::result::Result::Ok(std::option::Option::None),
|
|
};
|
|
}
|
|
|
|
/// Requests cooperative cancellation of one exact backend-issued Backfill Job identity.
|
|
pub(crate) fn cancel_backfill(&self, job_id: &str) -> ksp_core_lib::Result<crate::BackfillCancelResponseDto> {
|
|
let cancelled = self.backfill_runs.cancel(job_id);
|
|
let cancelled = match cancelled {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
domain = crate::TRACING_DOMAIN_RUN,
|
|
accepted = cancelled.accepted,
|
|
job_id = cancelled.job_id.as_str(),
|
|
state = cancelled.state.as_str(),
|
|
"processed Backfill Desk cooperative cancellation request"
|
|
);
|
|
return std::result::Result::Ok(cancelled);
|
|
}
|
|
|
|
/// Requests best-effort cooperative cancellation of the active run during application shutdown.
|
|
pub(crate) fn cancel_active_backfill_for_shutdown(&self) -> ksp_core_lib::Result<bool> {
|
|
return self.backfill_runs.cancel_active_for_shutdown();
|
|
}
|
|
|
|
/// 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();
|
|
}
|
|
|
|
/// Explicitly closes the retained Store runtime when application shutdown begins.
|
|
pub(crate) async fn close_store(&self) -> ksp_core_lib::Result<()> {
|
|
return self.store_startup.close().await;
|
|
}
|
|
|
|
/// Returns the resolved splash timings captured during application bootstrap.
|
|
#[must_use]
|
|
pub(crate) const fn splash_settings(&self) -> crate::SplashSettings {
|
|
return self.splash_settings;
|
|
}
|
|
|
|
/// Marks the one-shot splash lifecycle as started and reports whether this caller won the transition.
|
|
pub(crate) fn begin_splash_sequence(&self) -> bool {
|
|
return self
|
|
.splash_sequence_started
|
|
.compare_exchange(false, true, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire)
|
|
.is_ok();
|
|
}
|
|
}
|
|
|
|
struct LoggingRuntimeState {
|
|
guard: ksp_logging_lib::LoggingGuard,
|
|
active_profile_id: std::option::Option<String>,
|
|
fallback_active: bool,
|
|
startup_diagnostic: std::option::Option<crate::CommandErrorDto>,
|
|
}
|