v0.3.15-pre.009

crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 44
+// version: 45
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -32,6 +32,9 @@ const RAW_TRANSACTION_INGEST_STANDARD_BLOCK_PROTOCOL: &str = "solana_ws";
 const RAW_TRANSACTION_INGEST_STANDARD_LOGS_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_logs.filter.v1\0";
 const RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_PROTOCOL: &str = "solana_ws_http";
 const RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_logs_http.source_key.v1\0";
+const RAW_TRANSACTION_INGEST_YELLOWSTONE_BLOCK_HTTP_ACQUISITION_METHOD: &str = "block_subscribe_get_block";
+const RAW_TRANSACTION_INGEST_YELLOWSTONE_BLOCK_RECONCILIATION_ATTEMPTS: u8 = 4;
+const RAW_TRANSACTION_INGEST_YELLOWSTONE_BLOCK_RECONCILIATION_DELAY: std::time::Duration = std::time::Duration::from_millis(100);
 const RAW_TRANSACTION_INGEST_YELLOWSTONE_COVERAGE_SCOPE_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.coverage_scope.v1\0";
 const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0";
 const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL: &str = "yellowstone_http";
@@ -371,10 +374,15 @@ impl RawTransactionIngestLiveSource {
                     std::result::Result::Ok(value) => value,
                     std::result::Result::Err(error) => return std::result::Result::Err(error),
                 };
+                let known_reference_hydration =
+                    match http_role_supports_rpc_method(&source.http_pool, &source.hydration_role, "getTransaction", source.network.as_str()) {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => return std::result::Result::Err(error),
+                    };
                 let reference_bearing =
                     source.subscribe_request.transaction_filter_count() > 0 || source.subscribe_request.transaction_status_filter_count() > 0;
                 let live_block_material = source.subscribe_request.block_filter_count() > 0;
-                (reference_bearing, live_block_material, true, http_block_scan, true)
+                (reference_bearing, live_block_material, true, http_block_scan, known_reference_hydration)
             },
         };
         let block_material = live_block_material || http_block_scan;
@@ -404,7 +412,11 @@ impl RawTransactionIngestLiveSource {
     }
 
     fn uses_hydration(&self) -> bool {
-        return matches!(self, Self::HeliusTransaction(_) | Self::StandardLogs(_) | Self::Yellowstone(_));
+        return match self {
+            Self::HeliusTransaction(_) | Self::StandardLogs(_) => true,
+            Self::Yellowstone(source) => source.uses_transaction_hydration(),
+            Self::HttpBlockPolling(_) | Self::StandardBlock(_) => false,
+        };
     }
 
     async fn run(
@@ -975,6 +987,12 @@ struct RawTransactionIngestHydrationContext {
     source_key_domain: &'static [u8],
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RawTransactionIngestYellowstoneMode {
+    TransactionHydration,
+    BlockHydration,
+}
+
 /// 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 deterministic source-composition
@@ -984,6 +1002,7 @@ pub struct RawTransactionIngestYellowstoneSource {
     subscribe_request: ksp_onchain_transport_lib::YellowstoneSubscribeRequest,
     http_pool: ksp_onchain_transport_lib::HttpTransportPool,
     hydration_role: ksp_onchain_transport_lib::HttpRoleName,
+    mode: RawTransactionIngestYellowstoneMode,
     network: ksp_store_lib::RawNetworkId,
     route: RawTransactionIngestSourceRoute,
     source_key: [u8; 32],
@@ -1023,7 +1042,19 @@ impl crate::RawTransactionIngestYellowstoneSource {
             std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_endpoint_unrepresentable")),
         };
         let route = RawTransactionIngestSourceRoute { endpoint_id, provider };
-        let method = match ksp_onchain_transport_lib::find_http_rpc_method("getTransaction") {
+        let mode = if subscribe_request.block_filter_count() > 0
+            && subscribe_request.transaction_filter_count() == 0
+            && subscribe_request.transaction_status_filter_count() == 0
+        {
+            RawTransactionIngestYellowstoneMode::BlockHydration
+        } else {
+            RawTransactionIngestYellowstoneMode::TransactionHydration
+        };
+        let method_name = match mode {
+            RawTransactionIngestYellowstoneMode::TransactionHydration => "getTransaction",
+            RawTransactionIngestYellowstoneMode::BlockHydration => "getBlock",
+        };
+        let method = match ksp_onchain_transport_lib::find_http_rpc_method(method_name) {
             std::option::Option::Some(value) => value,
             std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_method_missing")),
         };
@@ -1040,7 +1071,11 @@ impl crate::RawTransactionIngestYellowstoneSource {
             std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_identity_invalid")),
         };
         let source_key = yellowstone_live_source_key(&network, &route, subscribe_request.commitment(), &request_identity);
-        return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, network, route, source_key });
+        return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, mode, network, route, source_key });
+    }
+
+    fn uses_transaction_hydration(&self) -> bool {
+        return self.mode == RawTransactionIngestYellowstoneMode::TransactionHydration;
     }
 
     fn hydration_context(&self) -> RawTransactionIngestHydrationContext {
@@ -1070,6 +1105,9 @@ impl crate::RawTransactionIngestYellowstoneSource {
         processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
         shared: RawTransactionIngestSourceRuntimeShared,
     ) -> ksp_core_lib::Result<()> {
+        if self.mode == RawTransactionIngestYellowstoneMode::BlockHydration {
+            return self.run_block_hydration(settings, stop_receiver, admission_sender, processing_frontier_sender).await;
+        }
         let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
             shared;
         let opened = tokio::select! {
@@ -1197,6 +1235,119 @@ impl crate::RawTransactionIngestYellowstoneSource {
             },
         };
     }
+
+    async fn run_block_hydration(
+        self,
+        settings: crate::RawTransactionIngestSettings,
+        mut stop_receiver: tokio::sync::watch::Receiver<bool>,
+        admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
+        processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
+    ) -> ksp_core_lib::Result<()> {
+        let opened = tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => {
+                return std::result::Result::Ok(());
+            }
+            result = self.yellowstone_channel.open_standard_subscribe(self.subscribe_request.clone()) => result,
+        };
+        let mut session = match opened {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())),
+        };
+        let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender);
+        let snapshot_source = session.snapshot_source();
+        if let std::result::Result::Err(error) = processing_frontier.observe_session_snapshot(snapshot_source.current()) {
+            processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
+            let _closed = session.close().await;
+            return std::result::Result::Err(error);
+        }
+        let mut session_snapshot_source = std::option::Option::Some(snapshot_source);
+        let mut fault = std::option::Option::None;
+        loop {
+            if *stop_receiver.borrow() {
+                break;
+            }
+            tokio::select! {
+                biased;
+                _ = stop_receiver.changed() => {
+                    break;
+                }
+                source_snapshot = wait_yellowstone_session_snapshot(&mut session_snapshot_source) => {
+                    match source_snapshot {
+                        std::option::Option::Some(snapshot) => {
+                            if let std::result::Result::Err(error) = processing_frontier.observe_session_snapshot(snapshot) {
+                                fault = std::option::Option::Some(error);
+                                break;
+                            }
+                        },
+                        std::option::Option::None => session_snapshot_source = std::option::Option::None,
+                    }
+                }
+                update = session.next_update() => {
+                    let update = match update {
+                        std::result::Result::Ok(std::option::Option::Some(value)) => value,
+                        std::result::Result::Ok(std::option::Option::None) => {
+                            fault = std::option::Option::Some(crate::runtime_error("source.session_closed"));
+                            break;
+                        },
+                        std::result::Result::Err(error) => {
+                            fault = std::option::Option::Some(source_transport_error(error.code()));
+                            break;
+                        },
+                    };
+                    if let ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(value) = update {
+                        let fetched = fetch_yellowstone_block_ingresses(&self, &settings, value.slot(), &mut stop_receiver).await;
+                        let ingresses = match fetched {
+                            std::result::Result::Ok(std::option::Option::Some(value)) => value,
+                            std::result::Result::Ok(std::option::Option::None) => break,
+                            std::result::Result::Err(error) => {
+                                fault = std::option::Option::Some(error);
+                                break;
+                            },
+                        };
+                        for ingress in ingresses {
+                            let sent = tokio::select! {
+                                biased;
+                                _ = stop_receiver.changed() => {
+                                    break;
+                                }
+                                result = admission_sender.send(ingress) => result,
+                            };
+                            if sent.is_err() {
+                                if !*stop_receiver.borrow() {
+                                    fault = std::option::Option::Some(crate::runtime_error("source.admission_closed"));
+                                }
+                                break;
+                            }
+                        }
+                        if fault.is_some() || *stop_receiver.borrow() {
+                            break;
+                        }
+                        if let std::result::Result::Err(error) = processing_frontier.observe_settled(value.slot()) {
+                            fault = std::option::Option::Some(error);
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+        processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closing);
+        let closed = session.close().await;
+        if let std::option::Option::Some(error) = fault {
+            processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
+            return std::result::Result::Err(error);
+        }
+        return match closed {
+            std::result::Result::Ok(()) => {
+                processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closed);
+                std::result::Result::Ok(())
+            },
+            std::result::Result::Err(error) => {
+                processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
+                std::result::Result::Err(source_transport_error(error.code()))
+            },
+        };
+    }
 }
 
 /// Validated Helius `transactionSubscribe` plus HTTP hydration source owned by the continuous RAW transaction ingest Worker.
@@ -3198,6 +3349,125 @@ fn validate_http_block_polling_discovery(next_scan_slot: u64, discovered: &[u64]
     return std::result::Result::Ok(());
 }
 
+async fn fetch_yellowstone_block_ingresses(
+    source: &crate::RawTransactionIngestYellowstoneSource,
+    settings: &crate::RawTransactionIngestSettings,
+    slot: u64,
+    stop_receiver: &mut tokio::sync::watch::Receiver<bool>,
+) -> ksp_core_lib::Result<std::option::Option<std::vec::Vec<crate::RawTransactionIngress>>> {
+    let commitment = match source.subscribe_request.commitment() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.yellowstone_commitment_missing")),
+    };
+    let config = ksp_onchain_transport_lib::SolanaGetBlockConfig::new(
+        std::option::Option::Some(commitment),
+        std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64),
+        std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionDetails::Full),
+        std::option::Option::Some(1),
+        std::option::Option::Some(false),
+    );
+    let mut attempt = 0_u8;
+    let observed = loop {
+        attempt = attempt.saturating_add(1);
+        let request = tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => {
+                return std::result::Result::Ok(std::option::Option::None);
+            }
+            result = source.http_pool.get_block_observed(&source.hydration_role, slot, std::option::Option::Some(&config)) => result,
+        };
+        match request {
+            std::result::Result::Ok(value) if value.value().is_some() => break value,
+            std::result::Result::Ok(_) if attempt < RAW_TRANSACTION_INGEST_YELLOWSTONE_BLOCK_RECONCILIATION_ATTEMPTS => {},
+            std::result::Result::Ok(_) => {
+                return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_not_available"));
+            },
+            std::result::Result::Err(_) if attempt < RAW_TRANSACTION_INGEST_YELLOWSTONE_BLOCK_RECONCILIATION_ATTEMPTS => {},
+            std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())),
+        }
+        tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => {
+                return std::result::Result::Ok(std::option::Option::None);
+            }
+            _ = tokio::time::sleep(RAW_TRANSACTION_INGEST_YELLOWSTONE_BLOCK_RECONCILIATION_DELAY) => {},
+        }
+    };
+    let block = match observed.value() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_not_available")),
+    };
+    let received_at = match current_raw_timestamp() {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let transactions = match block.transactions() {
+        ksp_onchain_transport_lib::SolanaWireField::Value(value) => value,
+        ksp_onchain_transport_lib::SolanaWireField::Omitted | ksp_onchain_transport_lib::SolanaWireField::Null => {
+            return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_transactions_missing"));
+        },
+    };
+    let provenance = build_yellowstone_block_provenance(source, settings, observed.endpoint_name(), observed.provider(), received_at);
+    let provenance = match provenance {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let mut ingresses = std::vec::Vec::with_capacity(transactions.len());
+    for (position, transaction) in transactions.iter().enumerate() {
+        let material = build_http_block_polling_material_from_transaction(&source.network, slot, block.block_time(), transaction, position);
+        let material = match material {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        ingresses.push(crate::RawTransactionIngress {
+            material,
+            network: source.network.clone(),
+            provenance: provenance.clone(),
+            source_key: source.source_key,
+        });
+    }
+    return std::result::Result::Ok(std::option::Option::Some(ingresses));
+}
+
+fn build_yellowstone_block_provenance(
+    source: &crate::RawTransactionIngestYellowstoneSource,
+    settings: &crate::RawTransactionIngestSettings,
+    endpoint_name: &str,
+    provider_name: &ksp_onchain_transport_lib::HttpProviderName,
+    received_at: ksp_store_lib::RawTimestamp,
+) -> ksp_core_lib::Result<ksp_store_lib::RawAcquisitionProvenance> {
+    let (provider, endpoint_id) = match composite_provenance_codes(&source.route, "ys", provider_name.as_str(), endpoint_name) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let protocol = match ksp_store_lib::RawProvenanceCode::new(RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_protocol_unrepresentable")),
+    };
+    let acquisition_method = match ksp_store_lib::RawProvenanceCode::new(RAW_TRANSACTION_INGEST_YELLOWSTONE_BLOCK_HTTP_ACQUISITION_METHOD) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_method_unrepresentable")),
+    };
+    let capture_session = match ksp_store_lib::RawProvenanceCode::new(settings.worker_id().as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_capture_session_unrepresentable")),
+    };
+    let commitment = match source.subscribe_request.commitment() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.yellowstone_commitment_missing")),
+    };
+    let commitment = match ksp_store_lib::RawProvenanceCode::new(commitment.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_commitment_unrepresentable")),
+    };
+    return std::result::Result::Ok(
+        ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, acquisition_method, ksp_store_lib::RawAcquisitionOrigin::Live, received_at)
+            .with_endpoint_id(endpoint_id)
+            .with_commitment(commitment)
+            .with_capture_session_id(capture_session),
+    );
+}
+
 fn project_http_block_polling_ingresses(
     source: &crate::RawTransactionIngestHttpBlockPollingSource,
     settings: &crate::RawTransactionIngestSettings,