v0.3.7-pre.005
This commit is contained in:
220
crates/ksp-app-backfill-desk/src/store_runtime.rs
Normal file
220
crates/ksp-app-backfill-desk/src/store_runtime.rs
Normal file
@@ -0,0 +1,220 @@
|
||||
// file: crates/ksp-app-backfill-desk/src/store_runtime.rs
|
||||
// version: 1
|
||||
|
||||
//! Composite-selected Store readiness and shutdown lifecycle owned by Backfill Desk.
|
||||
|
||||
/// Startup result for the Store layer, including safe diagnostics when the desktop shell remains available without a ready Store.
|
||||
pub(crate) struct StoreStartup {
|
||||
diagnostic: std::option::Option<crate::CommandErrorDto>,
|
||||
network_coherent: bool,
|
||||
runtime: std::option::Option<StoreRuntime>,
|
||||
store_network: std::option::Option<String>,
|
||||
}
|
||||
|
||||
impl StoreStartup {
|
||||
/// Applies Store readiness to the application-owned options DTO without exposing backend connection details.
|
||||
pub(crate) fn apply_to(&self, options: &mut crate::BackfillDeskOptionsDto) {
|
||||
options.network_coherent = self.network_coherent;
|
||||
options.store_diagnostic = self.diagnostic.clone();
|
||||
options.store_network = self.store_network.clone();
|
||||
options.store_ready = self.runtime.as_ref().is_some_and(StoreRuntime::health_ready);
|
||||
options.composition_ready = options.transport_ready && options.network_coherent && options.store_ready;
|
||||
}
|
||||
|
||||
/// Closes the retained Store runtime if startup reached the physical Store-open phase.
|
||||
pub(crate) async fn close(&self) -> ksp_core_lib::Result<()> {
|
||||
let runtime = self.runtime.as_ref();
|
||||
return match runtime {
|
||||
std::option::Option::Some(value) => value.close().await,
|
||||
std::option::Option::None => std::result::Result::Ok(()),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// Executable Store runtime retained after network coherence and bounded Store health checks.
|
||||
pub(crate) struct StoreRuntime {
|
||||
health_ready: bool,
|
||||
profile_id: String,
|
||||
store: std::sync::Mutex<std::option::Option<ksp_store_lib::Store>>,
|
||||
store_network: String,
|
||||
}
|
||||
|
||||
impl StoreRuntime {
|
||||
/// Returns whether the startup health probe proved the Store ready.
|
||||
#[must_use]
|
||||
pub(crate) const fn health_ready(&self) -> bool {
|
||||
return self.health_ready;
|
||||
}
|
||||
|
||||
/// Explicitly closes the Store exactly once through its backend-neutral facade.
|
||||
pub(crate) async fn close(&self) -> ksp_core_lib::Result<()> {
|
||||
let store = take_store(&self.store);
|
||||
let store = match store {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let store = match store {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => return std::result::Result::Ok(()),
|
||||
};
|
||||
let closed = store.close().await;
|
||||
return match closed {
|
||||
std::result::Result::Ok(()) => {
|
||||
ksp_logging_lib::debug!(
|
||||
target: crate::TRACING_TARGET,
|
||||
domain = crate::TRACING_DOMAIN_STORE,
|
||||
store_profile = self.profile_id.as_str(),
|
||||
store_network = self.store_network.as_str(),
|
||||
"closed Backfill Desk Store runtime"
|
||||
);
|
||||
std::result::Result::Ok(())
|
||||
},
|
||||
std::result::Result::Err(error) => std::result::Result::Err(
|
||||
ksp_core_lib::Error::new(crate::ERROR_CODE_STORE_SHUTDOWN_FAILED, "Backfill Desk Store shutdown failed").with_source(error),
|
||||
),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolves the composite Store target, proves Transport/Store network coherence before Store I/O, opens Store and captures one health probe.
|
||||
pub(crate) async fn initialize_store(
|
||||
management: &ksp_config_lib::ConfigManagement,
|
||||
transport_runtime: std::option::Option<&crate::TransportRuntime>,
|
||||
) -> StoreStartup {
|
||||
let environment = ksp_config_lib::ConfigEnvironment::load();
|
||||
let environment = match environment {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return unavailable_startup(std::option::Option::None, false, error),
|
||||
};
|
||||
let composite = crate::load_backfill_desk_composite(management);
|
||||
let composite = match composite {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return unavailable_startup(std::option::Option::None, false, error),
|
||||
};
|
||||
let profile = crate::required_composite_component_profile(&composite, crate::COMPOSITE_COMPONENT_ID_STORE, ksp_config_lib::FILE_ID_STD_STORE);
|
||||
let profile = match profile {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return unavailable_startup(std::option::Option::None, false, error),
|
||||
};
|
||||
let resolved = management.engine().resolve_store_config_profile(&profile, &environment);
|
||||
let resolved = match resolved {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return unavailable_startup(std::option::Option::None, false, error),
|
||||
};
|
||||
let profile_id = resolved.profile_id().to_owned();
|
||||
let store_network = resolved.settings().network().as_str().to_owned();
|
||||
let transport_network = transport_runtime.and_then(crate::TransportRuntime::coherent_network);
|
||||
let transport_network = match transport_network {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => {
|
||||
let error = ksp_core_lib::Error::new(
|
||||
crate::ERROR_CODE_STORE_NETWORK_UNAVAILABLE,
|
||||
"Backfill Desk cannot open Store before one coherent Transport network is available",
|
||||
);
|
||||
return unavailable_startup(std::option::Option::Some(store_network), false, error);
|
||||
},
|
||||
};
|
||||
let coherence = validate_network_coherence(store_network.as_str(), transport_network.as_str());
|
||||
if let std::result::Result::Err(error) = coherence {
|
||||
return unavailable_startup(std::option::Option::Some(store_network), false, error);
|
||||
}
|
||||
let store = ksp_store_lib::Store::open(resolved.into_settings()).await;
|
||||
let store = match store {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return unavailable_startup(std::option::Option::Some(store_network), true, error),
|
||||
};
|
||||
let health = store.health().await;
|
||||
let health_ready = store_health_ready(health.state());
|
||||
let diagnostic = if health_ready { std::option::Option::None } else { std::option::Option::Some(non_ready_health_diagnostic(&health)) };
|
||||
ksp_logging_lib::debug!(
|
||||
target: crate::TRACING_TARGET,
|
||||
domain = crate::TRACING_DOMAIN_STORE,
|
||||
store_profile = profile_id.as_str(),
|
||||
store_network = store_network.as_str(),
|
||||
store_ready = health_ready,
|
||||
pending_migration_count = health.pending_migration_count(),
|
||||
"initialized Backfill Desk Store readiness from composite-managed configuration"
|
||||
);
|
||||
return StoreStartup {
|
||||
diagnostic,
|
||||
network_coherent: true,
|
||||
runtime: std::option::Option::Some(StoreRuntime {
|
||||
health_ready,
|
||||
profile_id,
|
||||
store: std::sync::Mutex::new(std::option::Option::Some(store)),
|
||||
store_network: store_network.clone(),
|
||||
}),
|
||||
store_network: std::option::Option::Some(store_network),
|
||||
};
|
||||
}
|
||||
|
||||
fn non_ready_health_diagnostic(health: &ksp_store_lib::StoreHealthSnapshot) -> crate::CommandErrorDto {
|
||||
let code = health.last_error_code();
|
||||
return match code {
|
||||
std::option::Option::Some(value) => crate::CommandErrorDto {
|
||||
code: value.code().to_owned(),
|
||||
domain: value.domain().to_owned(),
|
||||
message: "Store health probe did not prove readiness".to_owned(),
|
||||
},
|
||||
std::option::Option::None => crate::CommandErrorDto {
|
||||
code: "store_not_ready".to_owned(),
|
||||
domain: "backfill_desk".to_owned(),
|
||||
message: "Store health probe did not prove readiness".to_owned(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
fn store_health_ready(state: ksp_store_lib::StoreHealthState) -> bool {
|
||||
return match state {
|
||||
ksp_store_lib::StoreHealthState::Ready => true,
|
||||
ksp_store_lib::StoreHealthState::NotReady => false,
|
||||
_ => false,
|
||||
};
|
||||
}
|
||||
|
||||
fn take_store(store: &std::sync::Mutex<std::option::Option<ksp_store_lib::Store>>) -> ksp_core_lib::Result<std::option::Option<ksp_store_lib::Store>> {
|
||||
let locked = store.lock();
|
||||
let mut locked = match locked {
|
||||
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 Store runtime state lock is poisoned",
|
||||
));
|
||||
},
|
||||
};
|
||||
return std::result::Result::Ok(locked.take());
|
||||
}
|
||||
|
||||
fn unavailable_startup(store_network: std::option::Option<String>, network_coherent: bool, error: ksp_core_lib::Error) -> StoreStartup {
|
||||
let diagnostic = crate::CommandErrorDto::from_error(&error);
|
||||
ksp_logging_lib::warn!(
|
||||
target: crate::TRACING_TARGET,
|
||||
domain = crate::TRACING_DOMAIN_STORE,
|
||||
error_domain = diagnostic.domain.as_str(),
|
||||
error_code = diagnostic.code.as_str(),
|
||||
network_coherent = network_coherent,
|
||||
"Backfill Desk Store readiness is unavailable; keeping desktop shell available"
|
||||
);
|
||||
return StoreStartup {
|
||||
diagnostic: std::option::Option::Some(diagnostic),
|
||||
network_coherent,
|
||||
runtime: std::option::Option::None,
|
||||
store_network,
|
||||
};
|
||||
}
|
||||
|
||||
fn validate_network_coherence(store_network: &str, transport_network: &str) -> ksp_core_lib::Result<()> {
|
||||
if store_network == transport_network {
|
||||
return std::result::Result::Ok(());
|
||||
}
|
||||
return std::result::Result::Err(
|
||||
ksp_core_lib::Error::new(crate::ERROR_CODE_STORE_NETWORK_MISMATCH, "Backfill Desk Store and Transport networks do not match")
|
||||
.with_context("store_network", store_network)
|
||||
.with_context("transport_network", transport_network),
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "../unit_tests/store_runtime.rs"]
|
||||
mod tests;
|
||||
Reference in New Issue
Block a user