// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs // version: 2 use sha2::Digest; // rust-rules: trait-import const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0"; #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum RawTransactionIngestSourceFamily { Transaction, TransactionStatus, } #[derive(Clone, Debug, Eq, PartialEq)] struct RawTransactionIngestSourceRoute { endpoint_id: ksp_store_lib::RawProvenanceCode, provider: ksp_store_lib::RawProvenanceCode, } #[derive(Clone, Copy, Eq, PartialEq)] struct RawTransactionIngestSourceTimestamp { nanos: u32, seconds: i64, } impl std::fmt::Debug for RawTransactionIngestSourceTimestamp { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter.debug_struct("RawTransactionIngestSourceTimestamp").field("nanos", &self.nanos).field("seconds", &self.seconds).finish(); } } #[derive(Clone, Eq, PartialEq)] struct RawTransactionIngestSourceSignal { created_at: std::option::Option, family: RawTransactionIngestSourceFamily, matched_filter_count: usize, matched_filter_fingerprint: [u8; 32], network: ksp_store_lib::RawNetworkId, route: RawTransactionIngestSourceRoute, signature: ksp_store_lib::RawTransactionSignature, slot: u64, transaction_index: std::option::Option, } impl std::fmt::Debug for RawTransactionIngestSourceSignal { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter .debug_struct("RawTransactionIngestSourceSignal") .field("created_at", &self.created_at) .field("family", &self.family) .field("matched_filter_count", &self.matched_filter_count) .field("matched_filter_fingerprint_bytes", &self.matched_filter_fingerprint.len()) .field("network", &self.network) .field("route", &self.route) .field("signature_bytes", &self.signature.as_bytes().len()) .field("slot", &self.slot) .field("has_transaction_index", &self.transaction_index.is_some()) .finish_non_exhaustive(); } } trait RawTransactionIngestYellowstoneSignalView { fn created_at(&self) -> std::option::Option; fn family(&self) -> RawTransactionIngestSourceFamily; fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]; fn index(&self) -> u64; fn signature(&self) -> ksp_onchain_transport_lib::YellowstoneTransactionSignature; fn slot(&self) -> u64; } impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionUpdate { fn created_at(&self) -> std::option::Option { return ksp_onchain_transport_lib::YellowstoneTransactionUpdate::created_at(self); } fn family(&self) -> RawTransactionIngestSourceFamily { return RawTransactionIngestSourceFamily::Transaction; } fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName] { return ksp_onchain_transport_lib::YellowstoneTransactionUpdate::filters(self); } fn index(&self) -> u64 { return ksp_onchain_transport_lib::YellowstoneTransactionUpdate::transaction(self).index(); } fn signature(&self) -> ksp_onchain_transport_lib::YellowstoneTransactionSignature { return ksp_onchain_transport_lib::YellowstoneTransactionUpdate::transaction(self).signature(); } fn slot(&self) -> u64 { return ksp_onchain_transport_lib::YellowstoneTransactionUpdate::slot(self); } } impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate { fn created_at(&self) -> std::option::Option { return ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate::created_at(self); } fn family(&self) -> RawTransactionIngestSourceFamily { return RawTransactionIngestSourceFamily::TransactionStatus; } fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName] { return ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate::filters(self); } fn index(&self) -> u64 { return ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate::index(self); } fn signature(&self) -> ksp_onchain_transport_lib::YellowstoneTransactionSignature { return ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate::signature(self); } fn slot(&self) -> u64 { return ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate::slot(self); } } impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)> for RawTransactionIngestSourceSignal { fn from(value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)) -> Self { return project_yellowstone_signal(value.0, value.1); } } impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)> for RawTransactionIngestSourceSignal { fn from(value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)) -> Self { return project_yellowstone_signal(value.0, value.1); } } /// Validated Yellowstone plus HTTP runtime source owned by the continuous RAW transaction ingest Worker. /// /// The Transport-owned channel, subscribe request, HTTP pool and hydration role remain private. Construction validates only deterministic source-composition /// invariants and performs no network I/O. pub struct RawTransactionIngestYellowstoneSource { yellowstone_channel: ksp_onchain_transport_lib::YellowstoneGrpcChannel, subscribe_request: ksp_onchain_transport_lib::YellowstoneSubscribeRequest, http_pool: ksp_onchain_transport_lib::HttpTransportPool, hydration_role: ksp_onchain_transport_lib::HttpRoleName, network: ksp_store_lib::RawNetworkId, route: RawTransactionIngestSourceRoute, } impl crate::RawTransactionIngestYellowstoneSource { /// Creates one validated Yellowstone source contract without opening a stream or issuing HTTP requests. pub fn new( yellowstone_channel: ksp_onchain_transport_lib::YellowstoneGrpcChannel, subscribe_request: ksp_onchain_transport_lib::YellowstoneSubscribeRequest, http_pool: ksp_onchain_transport_lib::HttpTransportPool, hydration_role: ksp_onchain_transport_lib::HttpRoleName, ) -> ksp_core_lib::Result { if subscribe_request.validate().is_err() { return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_request_invalid")); } if ingestion_filter_count(&subscribe_request) == 0 { return std::result::Result::Err(crate::runtime_error("runtime_resources.ingestion_filter_missing")); } match subscribe_request.commitment() { std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) | std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => {}, std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed) | std::option::Option::None => { return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_commitment_invalid")); }, } let network = match ksp_store_lib::RawNetworkId::new(yellowstone_channel.cluster().as_str()) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_network_unrepresentable")), }; let provider = match ksp_store_lib::RawProvenanceCode::new(yellowstone_channel.provider().as_str()) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_provider_unrepresentable")), }; let endpoint_id = match ksp_store_lib::RawProvenanceCode::new(yellowstone_channel.endpoint_name()) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_endpoint_unrepresentable")), }; let method = match ksp_onchain_transport_lib::find_http_rpc_method("getTransaction") { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_method_missing")), }; let compatible_http_routes = compatible_http_route_count(&http_pool, &hydration_role, method.request_kind(), network.as_str()); let compatible_http_routes = match compatible_http_routes { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if compatible_http_routes == 0 { return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_role_unsupported")); } let route = RawTransactionIngestSourceRoute { endpoint_id, provider }; return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, network, route }); } } impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let http_snapshot = self.http_pool.snapshot(); return formatter .debug_struct("RawTransactionIngestYellowstoneSource") .field("yellowstone_endpoint_name", &self.yellowstone_channel.endpoint_name()) .field("yellowstone_provider", &self.yellowstone_channel.provider().as_str()) .field("network", &self.network.as_str()) .field("transaction_filter_count", &self.subscribe_request.transaction_filter_count()) .field("transaction_status_filter_count", &self.subscribe_request.transaction_status_filter_count()) .field("block_filter_count", &self.subscribe_request.block_filter_count()) .field("commitment", &self.subscribe_request.commitment()) .field("has_from_slot", &self.subscribe_request.from_slot().is_some()) .field("hydration_role", &self.hydration_role.as_str()) .field("http_endpoint_count", &http_snapshot.endpoint_count()) .finish(); } } /// Caller-composed runtime resources accepted by the continuous RAW transaction ingest Worker. /// /// This first runtime-resource contract owns exactly one Yellowstone source. It deliberately does not expose a provider enum, source collection, callback, /// enqueue surface or lower-layer client escape hatch. pub struct RawTransactionIngestRuntimeResources { yellowstone_source: crate::RawTransactionIngestYellowstoneSource, } impl crate::RawTransactionIngestRuntimeResources { /// Owns the first validated productive-source contract for a future runtime start. #[must_use] pub fn new(yellowstone_source: crate::RawTransactionIngestYellowstoneSource) -> Self { return Self { yellowstone_source }; } /// Validates that caller-owned Worker settings target the same logical network as the composed Yellowstone/HTTP source. pub(crate) fn validate_network(&self, network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> { if &self.yellowstone_source.network != network { return std::result::Result::Err(crate::runtime_error("runtime_resources.worker_network_mismatch")); } return std::result::Result::Ok(()); } } impl std::fmt::Debug for crate::RawTransactionIngestRuntimeResources { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter.debug_struct("RawTransactionIngestRuntimeResources").field("yellowstone_source", &self.yellowstone_source).finish(); } } fn compatible_http_route_count( pool: &ksp_onchain_transport_lib::HttpTransportPool, hydration_role: &ksp_onchain_transport_lib::HttpRoleName, request_kind: &str, expected_cluster: &str, ) -> ksp_core_lib::Result { let snapshot = pool.snapshot(); let mut compatible = 0_usize; for endpoint in snapshot.endpoints() { if !endpoint.enabled() { continue; } for role in endpoint.roles() { if !role.enabled() || role.role() != hydration_role.as_str() { continue; } let supports_request = role.request_kinds().iter().any(|kind| return kind.as_str() == "*" || kind.as_str() == request_kind); if !supports_request { continue; } if endpoint.cluster() != expected_cluster { return std::result::Result::Err(crate::runtime_error("runtime_resources.transport_network_mismatch")); } compatible = compatible.saturating_add(1); } } return std::result::Result::Ok(compatible); } fn ingestion_filter_count(request: &ksp_onchain_transport_lib::YellowstoneSubscribeRequest) -> usize { return request.transaction_filter_count().saturating_add(request.transaction_status_filter_count()).saturating_add(request.block_filter_count()); } fn matched_filter_fingerprint(filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]) -> [u8; 32] { let mut names = filters.iter().map(ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::as_str).collect::>(); names.sort_unstable(); names.dedup(); let mut hasher = sha2::Sha256::new(); hasher.update(RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN); hasher.update((names.len() as u64).to_be_bytes()); for name in names { hasher.update((name.len() as u64).to_be_bytes()); hasher.update(name.as_bytes()); } return hasher.finalize().into(); } fn project_yellowstone_signal( source: &crate::RawTransactionIngestYellowstoneSource, update: &T, ) -> RawTransactionIngestSourceSignal { let signature = ksp_store_lib::RawTransactionSignature::new(*update.signature().as_bytes()); let created_at = update.created_at().map(|value| { return RawTransactionIngestSourceTimestamp { nanos: value.nanos(), seconds: value.seconds() }; }); return RawTransactionIngestSourceSignal { created_at, family: update.family(), matched_filter_count: update.filters().len(), matched_filter_fingerprint: matched_filter_fingerprint(update.filters()), network: source.network.clone(), route: source.route.clone(), signature, slot: update.slot(), transaction_index: std::option::Option::Some(update.index()), }; } #[cfg(test)] #[path = "../unit_tests/runtime_resources.rs"] mod tests;