khadhroony-solana-project/crates/ksp-app-backfill-desk/src/store_runtime.rs

// file: crates/ksp-app-backfill-desk/src/store_runtime.rs
// version: 2

//! 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;
    }

    /// Temporarily lends the ready Store facade to one admitted Backfill execution.
    pub(crate) fn take_for_run(&self) -> ksp_core_lib::Result<ksp_store_lib::Store> {
        let runtime = self.runtime.as_ref();
        let runtime = match runtime {
            std::option::Option::Some(value) if value.health_ready() => value,
            _ => {
                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 Store",
                ));
            },
        };
        return runtime.take_for_run();
    }

    /// Restores the Store facade after one Backfill execution reaches its terminal result.
    pub(crate) fn restore_after_run(&self, store: ksp_store_lib::Store) -> ksp_core_lib::Result<()> {
        let runtime = self.runtime.as_ref();
        let runtime = match runtime {
            std::option::Option::Some(value) => value,
            std::option::Option::None => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_INVALID,
                    "Backfill Desk cannot restore Store after runtime ownership disappeared",
                ));
            },
        };
        return runtime.restore_after_run(store);
    }

    /// 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;
    }

    /// Temporarily removes the Store from the application slot for one single-active Backfill execution.
    pub(crate) fn take_for_run(&self) -> ksp_core_lib::Result<ksp_store_lib::Store> {
        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),
        };
        return match store {
            std::option::Option::Some(value) => std::result::Result::Ok(value),
            std::option::Option::None => std::result::Result::Err(ksp_core_lib::Error::new(
                crate::ERROR_CODE_BACKFILL_RUN_ACTIVE,
                "Backfill Desk Store is already owned by an active Backfill run",
            )),
        };
    }

    /// Restores the Store after one terminal Backfill execution so a later campaign can be admitted.
    pub(crate) fn restore_after_run(&self, store: ksp_store_lib::Store) -> ksp_core_lib::Result<()> {
        let locked = self.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",
                ));
            },
        };
        if locked.is_some() {
            return std::result::Result::Err(ksp_core_lib::Error::new(
                crate::ERROR_CODE_APP_STATE_INVALID,
                "Backfill Desk cannot restore Store into an occupied runtime slot",
            ));
        }
        *locked = std::option::Option::Some(store);
        return std::result::Result::Ok(());
    }

    /// 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;