v0.3.15-pre.007

2026-09-13 17:36:06 +02:00
parent 487bcbf0c1
commit 0d9ec5c490
20 changed files with 1071 additions and 109 deletions
--- a/crates/ksp-app-raw-transaction-ingest-desk/src/route_start.rs
+++ b/crates/ksp-app-raw-transaction-ingest-desk/src/route_start.rs
@@ -0,0 +1,493 @@
+// file: crates/ksp-app-raw-transaction-ingest-desk/src/route_start.rs
+// version: 1
+
+//! Start-time Config revalidation and Transport-resource reconstruction without Worker launch.
+
+/// Revalidates one future Start request against the current inventory generation and reconstructs the exact Worker source contract without starting it.
+pub(crate) fn validate_route_start(
+    management: &ksp_config_lib::ConfigManagement,
+    current_generation: u32,
+    request: &crate::RawIngestRouteStartRequestDto,
+) -> ksp_core_lib::Result<crate::RawIngestRouteStartValidationDto> {
+    if current_generation == 0 || request.inventory_generation != current_generation {
+        return std::result::Result::Err(ksp_core_lib::Error::new(
+            crate::ERROR_CODE_ROUTE_START_STALE_INVENTORY,
+            "Raw Transaction Ingest Desk Start request uses a stale route inventory generation",
+        ));
+    }
+    let environment = ksp_config_lib::ConfigEnvironment::load();
+    let environment = match environment {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return route_start_error("Cannot resolve Start-time Config environment", error),
+    };
+    let inventory = crate::build_route_inventory_with_environment(management, &environment, current_generation);
+    let inventory = match inventory {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return route_start_error("Cannot rebuild Start-time route inventory", error),
+    };
+    let profile = inventory.profiles.iter().find(|profile| return profile.profile_id == request.profile_id);
+    let profile = match profile {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return route_unavailable_error(),
+    };
+    let route = profile.routes.iter().find(|route| return route.route_id == request.route_id);
+    let route = match route {
+        std::option::Option::Some(value) if value.selectable && value.state == crate::RawIngestRouteState::Configured => value,
+        std::option::Option::Some(_) | std::option::Option::None => return route_unavailable_error(),
+    };
+    let network = match route.network.as_deref() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return route_unavailable_error(),
+    };
+    let composite = crate::load_raw_transaction_ingest_desk_composite_profile(management, std::option::Option::Some(request.profile_id.as_str()));
+    let composite = match composite {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return route_start_error("Cannot resolve Start-time composite profile", error),
+    };
+    let store_profile = crate::required_composite_component_profile(&composite, crate::COMPOSITE_COMPONENT_ID_STORE, ksp_config_lib::FILE_ID_STD_STORE);
+    let store_profile = match store_profile {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return route_start_error("Cannot resolve Start-time Store profile", error),
+    };
+    let store = management.engine().resolve_store_config_profile(&store_profile, &environment);
+    let store = match store {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return route_start_error("Cannot resolve Start-time Store settings", error),
+    };
+    if store.settings().network().as_str() != network {
+        return std::result::Result::Err(ksp_core_lib::Error::new(
+            crate::ERROR_CODE_ROUTE_START_PREPARATION_FAILED,
+            "Raw Transaction Ingest Desk Start-time Store network no longer matches the selected route",
+        ));
+    }
+    let transport_profile = crate::required_composite_component_profile(
+        &composite,
+        crate::COMPOSITE_COMPONENT_ID_TRANSPORT,
+        ksp_config_lib::FILE_ID_STD_TRANSPORT,
+    );
+    let transport_profile = match transport_profile {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return route_start_error("Cannot resolve Start-time base Transport profile", error),
+    };
+    let transport = management.engine().resolve_transport_config_profile(&transport_profile, &environment);
+    let transport = match transport {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return route_start_error("Cannot resolve Start-time base Transport settings", error),
+    };
+    if !transport_matches_network(&transport, network) {
+        return std::result::Result::Err(ksp_core_lib::Error::new(
+            crate::ERROR_CODE_ROUTE_START_PREPARATION_FAILED,
+            "Raw Transaction Ingest Desk Start-time Transport network no longer matches the selected route",
+        ));
+    }
+    let commitment = request.commitment.into_transport();
+    let prepared = match request.route_id {
+        crate::RawIngestRouteId::YellowstoneHydrated => prepare_yellowstone(management, &environment, &composite, &transport, network, commitment),
+        crate::RawIngestRouteId::StandardLogsHydrated => prepare_standard_logs(&transport, network, commitment),
+        crate::RawIngestRouteId::StandardBlockDirect => prepare_standard_block(&transport, network, commitment),
+        crate::RawIngestRouteId::HeliusTransactionHydrated => prepare_helius(management, &environment, &composite, &transport, network, commitment),
+        crate::RawIngestRouteId::HttpBlockPolling => prepare_http_polling(&transport, network, commitment),
+    };
+    if let std::result::Result::Err(error) = prepared {
+        return route_start_error("Cannot reconstruct Start-time Transport resources", error);
+    }
+    ksp_logging_lib::debug!(
+        target: crate::TRACING_TARGET,
+        domain = crate::TRACING_DOMAIN_ROUTE_START,
+        profile_id = request.profile_id.as_str(),
+        network,
+        route_id = request.route_id.as_str(),
+        inventory_generation = request.inventory_generation,
+        commitment = request.commitment.as_str(),
+        "revalidated Start request and reconstructed exact Transport-owned Worker source contract without runtime I/O"
+    );
+    return std::result::Result::Ok(crate::RawIngestRouteStartValidationDto {
+        commitment: request.commitment,
+        inventory_generation: request.inventory_generation,
+        network: network.to_owned(),
+        profile_id: request.profile_id.clone(),
+        route_id: request.route_id,
+    });
+}
+
+fn prepare_standard_logs(
+    transport: &ksp_config_lib::ResolvedTransportConfig,
+    network: &str,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+) -> ksp_core_lib::Result<()> {
+    let endpoint = single_ws_endpoint(
+        transport,
+        network,
+        ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard,
+        ksp_onchain_transport_lib::WsSubscriptionKind::Logs,
+    );
+    let endpoint = match endpoint {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let role = single_http_role(transport.http_settings(), network, &["getTransaction"]);
+    let role = match role {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let pool = ksp_onchain_transport_lib::HttpTransportPool::new(transport.http_settings().clone());
+    let pool = match pool {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let source = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestStandardLogsSource::new(
+        endpoint,
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes,
+        commitment,
+        pool,
+        role,
+    );
+    let source = match source {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    std::mem::drop(source);
+    return std::result::Result::Ok(());
+}
+
+fn prepare_standard_block(
+    transport: &ksp_config_lib::ResolvedTransportConfig,
+    network: &str,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+) -> ksp_core_lib::Result<()> {
+    let endpoint = single_ws_endpoint(
+        transport,
+        network,
+        ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard,
+        ksp_onchain_transport_lib::WsSubscriptionKind::Block,
+    );
+    let endpoint = match endpoint {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let source = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestStandardBlockSource::new(
+        endpoint,
+        ksp_onchain_transport_lib::SolanaBlockSubscribeFilter::All,
+        commitment,
+    );
+    let source = match source {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    std::mem::drop(source);
+    return std::result::Result::Ok(());
+}
+
+fn prepare_helius(
+    management: &ksp_config_lib::ConfigManagement,
+    environment: &ksp_config_lib::ConfigEnvironment,
+    composite: &ksp_config_lib::ResolvedConfigComposite,
+    base_transport: &ksp_config_lib::ResolvedTransportConfig,
+    network: &str,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+) -> ksp_core_lib::Result<()> {
+    let transport = resolve_optional_transport(management, environment, composite, crate::COMPOSITE_COMPONENT_ID_TRANSPORT_HELIUS, network);
+    let transport = match transport {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let endpoint = single_ws_endpoint(
+        &transport,
+        network,
+        ksp_onchain_transport_lib::WsProtocolKind::HeliusLaserStream,
+        ksp_onchain_transport_lib::WsSubscriptionKind::HeliusTransaction,
+    );
+    let endpoint = match endpoint {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let role = single_http_role(base_transport.http_settings(), network, &["getTransaction"]);
+    let role = match role {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let pool = ksp_onchain_transport_lib::HttpTransportPool::new(base_transport.http_settings().clone());
+    let pool = match pool {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let filter = ksp_onchain_transport_lib::HeliusTransactionSubscribeFilter::new(
+        std::option::Option::None,
+        std::option::Option::None,
+        std::option::Option::None,
+        std::option::Option::None,
+        std::option::Option::None,
+        std::option::Option::None,
+        std::option::Option::None,
+    );
+    let source = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestHeliusTransactionSource::new(endpoint, filter, commitment, pool, role);
+    let source = match source {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    std::mem::drop(source);
+    return std::result::Result::Ok(());
+}
+
+fn prepare_yellowstone(
+    management: &ksp_config_lib::ConfigManagement,
+    environment: &ksp_config_lib::ConfigEnvironment,
+    composite: &ksp_config_lib::ResolvedConfigComposite,
+    base_transport: &ksp_config_lib::ResolvedTransportConfig,
+    network: &str,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+) -> ksp_core_lib::Result<()> {
+    let transport = resolve_optional_transport(management, environment, composite, crate::COMPOSITE_COMPONENT_ID_TRANSPORT_YELLOWSTONE, network);
+    let transport = match transport {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let endpoint = single_grpc_endpoint(&transport, network);
+    let endpoint = match endpoint {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let channel = ksp_onchain_transport_lib::YellowstoneGrpcChannel::prepare(&endpoint);
+    let channel = match channel {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let filter_name = ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::new("raw-ingest-all-transactions");
+    let filter_name = match filter_name {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let mut request = ksp_onchain_transport_lib::YellowstoneSubscribeRequest::new();
+    let inserted = request.insert_transaction_filter(filter_name, ksp_onchain_transport_lib::YellowstoneSubscribeTransactionFilter::new());
+    if let std::result::Result::Err(error) = inserted {
+        return std::result::Result::Err(error);
+    }
+    request.set_commitment(std::option::Option::Some(commitment));
+    let role = single_http_role(base_transport.http_settings(), network, &["getTransaction"]);
+    let role = match role {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let pool = ksp_onchain_transport_lib::HttpTransportPool::new(base_transport.http_settings().clone());
+    let pool = match pool {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let source = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestYellowstoneSource::new(channel, request, pool, role);
+    let source = match source {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    std::mem::drop(source);
+    return std::result::Result::Ok(());
+}
+
+fn prepare_http_polling(
+    transport: &ksp_config_lib::ResolvedTransportConfig,
+    network: &str,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+) -> ksp_core_lib::Result<()> {
+    let role = single_http_role(transport.http_settings(), network, &["getSlot", "getBlocksWithLimit", "getBlock"]);
+    let role = match role {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let pool = ksp_onchain_transport_lib::HttpTransportPool::new(transport.http_settings().clone());
+    let pool = match pool {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let source = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestHttpBlockPollingSource::new(pool, role, commitment);
+    let source = match source {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    std::mem::drop(source);
+    return std::result::Result::Ok(());
+}
+
+fn resolve_optional_transport(
+    management: &ksp_config_lib::ConfigManagement,
+    environment: &ksp_config_lib::ConfigEnvironment,
+    composite: &ksp_config_lib::ResolvedConfigComposite,
+    component_id: &str,
+    network: &str,
+) -> ksp_core_lib::Result<ksp_config_lib::ResolvedTransportConfig> {
+    let component = composite.component(component_id);
+    let component = match component {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => {
+            return std::result::Result::Err(ksp_core_lib::Error::new(
+                crate::ERROR_CODE_ROUTE_START_PREPARATION_FAILED,
+                "Raw Transaction Ingest Desk selected route has no declared Transport source",
+            ));
+        },
+    };
+    if component.resolved().file_id().as_str() != ksp_config_lib::FILE_ID_STD_TRANSPORT {
+        return std::result::Result::Err(ksp_core_lib::Error::new(
+            crate::ERROR_CODE_ROUTE_START_PREPARATION_FAILED,
+            "Raw Transaction Ingest Desk selected route references an unexpected Transport document",
+        ));
+    }
+    let resolved = management.engine().resolve_transport_config_profile(component.resolved(), environment);
+    let resolved = match resolved {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if !transport_matches_network(&resolved, network) {
+        return std::result::Result::Err(ksp_core_lib::Error::new(
+            crate::ERROR_CODE_ROUTE_START_PREPARATION_FAILED,
+            "Raw Transaction Ingest Desk provider Transport network no longer matches the selected route",
+        ));
+    }
+    return std::result::Result::Ok(resolved);
+}
+
+fn single_ws_endpoint(
+    transport: &ksp_config_lib::ResolvedTransportConfig,
+    network: &str,
+    protocol: ksp_onchain_transport_lib::WsProtocolKind,
+    capability: ksp_onchain_transport_lib::WsSubscriptionKind,
+) -> ksp_core_lib::Result<ksp_onchain_transport_lib::WsEndpointSettings> {
+    let settings = match transport.ws_settings() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return ambiguous_resource_error("Start-time route requires one WebSocket endpoint but none is configured"),
+    };
+    let matching = settings
+        .endpoints()
+        .iter()
+        .filter(|endpoint| {
+            return endpoint.enabled()
+                && endpoint.cluster().as_str() == network
+                && endpoint.protocol() == protocol
+                && endpoint.supports_subscription(capability);
+        })
+        .collect::<std::vec::Vec<_>>();
+    if matching.len() != 1 {
+        return ambiguous_resource_error("Start-time route requires exactly one matching WebSocket endpoint in its selected Transport profile");
+    }
+    return std::result::Result::Ok(matching[0].clone());
+}
+
+fn single_grpc_endpoint(
+    transport: &ksp_config_lib::ResolvedTransportConfig,
+    network: &str,
+) -> ksp_core_lib::Result<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
+    let settings = match transport.grpc_settings() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return ambiguous_resource_error("Start-time route requires one Yellowstone gRPC endpoint but none is configured"),
+    };
+    let matching = settings
+        .endpoints()
+        .iter()
+        .filter(|endpoint| return endpoint.enabled() && endpoint.cluster().as_str() == network)
+        .collect::<std::vec::Vec<_>>();
+    if matching.len() != 1 {
+        return ambiguous_resource_error("Start-time route requires exactly one matching Yellowstone gRPC endpoint in its selected Transport profile");
+    }
+    return std::result::Result::Ok(matching[0].clone());
+}
+
+fn single_http_role(
+    settings: &ksp_onchain_transport_lib::HttpTransportSettings,
+    network: &str,
+    methods: &[&'static str],
+) -> ksp_core_lib::Result<ksp_onchain_transport_lib::HttpRoleName> {
+    let mut request_kinds = std::vec::Vec::<&'static str>::with_capacity(methods.len());
+    for method_name in methods {
+        let descriptor = ksp_onchain_transport_lib::find_http_rpc_method(method_name);
+        let descriptor = match descriptor {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return ambiguous_resource_error("Start-time route references an unknown HTTP method requirement"),
+        };
+        request_kinds.push(descriptor.request_kind());
+    }
+    let mut roles = std::collections::BTreeSet::<String>::new();
+    for endpoint in settings.endpoints() {
+        if !endpoint.enabled() || endpoint.cluster().as_str() != network {
+            continue;
+        }
+        for role in endpoint.roles() {
+            if role.enabled() {
+                roles.insert(role.role().as_str().to_owned());
+            }
+        }
+    }
+    let matching = roles
+        .into_iter()
+        .filter(|role_name| {
+            return request_kinds
+                .iter()
+                .all(|request_kind| return http_role_supports_request_kind(settings, network, role_name.as_str(), request_kind));
+        })
+        .collect::<std::vec::Vec<_>>();
+    if matching.len() != 1 {
+        return ambiguous_resource_error("Start-time route requires exactly one logical HTTP role satisfying its complete method set");
+    }
+    return std::result::Result::Ok(ksp_onchain_transport_lib::HttpRoleName::new(matching[0].clone()));
+}
+
+fn http_role_supports_request_kind(
+    settings: &ksp_onchain_transport_lib::HttpTransportSettings,
+    network: &str,
+    role_name: &str,
+    request_kind: &str,
+) -> bool {
+    for endpoint in settings.endpoints() {
+        if !endpoint.enabled() || endpoint.cluster().as_str() != network {
+            continue;
+        }
+        for role in endpoint.roles() {
+            if !role.enabled() || role.role().as_str() != role_name {
+                continue;
+            }
+            if role.request_kinds().iter().any(|kind| return kind.is_wildcard() || kind.as_str() == request_kind) {
+                return true;
+            }
+        }
+    }
+    return false;
+}
+
+fn transport_matches_network(transport: &ksp_config_lib::ResolvedTransportConfig, network: &str) -> bool {
+    let mut networks = std::collections::BTreeSet::<String>::new();
+    for endpoint in transport.http_settings().endpoints() {
+        if endpoint.enabled() {
+            networks.insert(endpoint.cluster().as_str().to_owned());
+        }
+    }
+    if let std::option::Option::Some(settings) = transport.ws_settings() {
+        for endpoint in settings.endpoints() {
+            if endpoint.enabled() {
+                networks.insert(endpoint.cluster().as_str().to_owned());
+            }
+        }
+    }
+    if let std::option::Option::Some(settings) = transport.grpc_settings() {
+        for endpoint in settings.endpoints() {
+            if endpoint.enabled() {
+                networks.insert(endpoint.cluster().as_str().to_owned());
+            }
+        }
+    }
+    return networks.len() == 1 && networks.contains(network);
+}
+
+fn route_unavailable_error<T>() -> ksp_core_lib::Result<T> {
+    return std::result::Result::Err(ksp_core_lib::Error::new(
+        crate::ERROR_CODE_ROUTE_START_UNAVAILABLE,
+        "Raw Transaction Ingest Desk selected route is no longer Configured in the current Start-time inventory",
+    ));
+}
+
+fn ambiguous_resource_error<T>(message: &'static str) -> ksp_core_lib::Result<T> {
+    return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_ROUTE_START_PREPARATION_FAILED, message));
+}
+
+fn route_start_error<T>(message: &'static str, source: ksp_core_lib::Error) -> ksp_core_lib::Result<T> {
+    return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_ROUTE_START_PREPARATION_FAILED, message).with_source(source));
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/route_start.rs"]
+mod tests;