v0.3.12-pre.004

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

@@ -1,11 +1,15 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 3
+// version: 4
 
 #[cfg(test)]
 use sha2::Digest; // rust-rules: trait-import
 
 #[cfg(test)]
 const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0";
+#[cfg(test)]
+const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL: &str = "yellowstone_http";
+#[cfg(test)]
+const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone_http.source_key.v1\0";
 
 #[cfg(test)]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -41,6 +45,7 @@ struct RawTransactionIngestSourceSignal {
     family: RawTransactionIngestSourceFamily,
     matched_filter_count: usize,
     matched_filter_fingerprint: [u8; 32],
+    matched_filter_id: std::option::Option<ksp_store_lib::RawProvenanceCode>,
     network: ksp_store_lib::RawNetworkId,
     route: RawTransactionIngestSourceRoute,
     signature: ksp_store_lib::RawTransactionSignature,
@@ -57,6 +62,7 @@ impl std::fmt::Debug for RawTransactionIngestSourceSignal {
             .field("family", &self.family)
             .field("matched_filter_count", &self.matched_filter_count)
             .field("matched_filter_fingerprint_bytes", &self.matched_filter_fingerprint.len())
+            .field("has_direct_filter_id", &self.matched_filter_id.is_some())
             .field("network", &self.network)
             .field("route", &self.route)
             .field("signature_bytes", &self.signature.as_bytes().len())
@@ -199,11 +205,12 @@ impl crate::RawTransactionIngestYellowstoneSource {
             std::result::Result::Ok(value) => value,
             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") {
             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 = compatible_http_route_count(&http_pool, &hydration_role, method.request_kind(), network.as_str(), &route);
         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),
@@ -211,7 +218,6 @@ impl crate::RawTransactionIngestYellowstoneSource {
         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 });
     }
 }
@@ -271,6 +277,7 @@ fn compatible_http_route_count(
     hydration_role: &ksp_onchain_transport_lib::HttpRoleName,
     request_kind: &str,
     expected_cluster: &str,
+    yellowstone_route: &RawTransactionIngestSourceRoute,
 ) -> ksp_core_lib::Result<usize> {
     let snapshot = pool.snapshot();
     let mut compatible = 0_usize;
@@ -289,16 +296,56 @@ fn compatible_http_route_count(
             if endpoint.cluster() != expected_cluster {
                 return std::result::Result::Err(crate::runtime_error("runtime_resources.transport_network_mismatch"));
             }
+            let composite = composite_provenance_codes(yellowstone_route, endpoint.provider(), endpoint.name());
+            if let std::result::Result::Err(error) = composite {
+                return std::result::Result::Err(error);
+            }
             compatible = compatible.saturating_add(1);
         }
     }
     return std::result::Result::Ok(compatible);
 }
 
+fn composite_provenance_codes(
+    yellowstone_route: &RawTransactionIngestSourceRoute,
+    http_provider: &str,
+    http_endpoint: &str,
+) -> ksp_core_lib::Result<(ksp_store_lib::RawProvenanceCode, ksp_store_lib::RawProvenanceCode)> {
+    let provider = std::format!("ys.{}:http.{}", yellowstone_route.provider.as_str(), http_provider);
+    let provider = match ksp_store_lib::RawProvenanceCode::new(provider) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.composite_provider_unrepresentable"));
+        },
+    };
+    let endpoint = std::format!("ys.{}:http.{}", yellowstone_route.endpoint_id.as_str(), http_endpoint);
+    let endpoint = match ksp_store_lib::RawProvenanceCode::new(endpoint) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.composite_endpoint_unrepresentable"));
+        },
+    };
+    return std::result::Result::Ok((provider, endpoint));
+}
+
 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());
 }
 
+#[cfg(test)]
+fn matched_filter_direct_id(filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]) -> std::option::Option<ksp_store_lib::RawProvenanceCode> {
+    let mut names = filters.iter().map(ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::as_str).collect::<std::vec::Vec<_>>();
+    names.sort_unstable();
+    names.dedup();
+    if names.len() != 1 {
+        return std::option::Option::None;
+    }
+    return match ksp_store_lib::RawProvenanceCode::new(names[0]) {
+        std::result::Result::Ok(value) => std::option::Option::Some(value),
+        std::result::Result::Err(_) => std::option::Option::None,
+    };
+}
+
 #[cfg(test)]
 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::<std::vec::Vec<_>>();
@@ -328,6 +375,7 @@ fn project_yellowstone_signal<T: RawTransactionIngestYellowstoneSignalView>(
         family: update.family(),
         matched_filter_count: update.filters().len(),
         matched_filter_fingerprint: matched_filter_fingerprint(update.filters()),
+        matched_filter_id: matched_filter_direct_id(update.filters()),
         network: source.network.clone(),
         route: source.route.clone(),
         signature,
@@ -336,6 +384,264 @@ fn project_yellowstone_signal<T: RawTransactionIngestYellowstoneSignalView>(
     };
 }
 
+#[cfg(test)]
+enum RawTransactionIngestHydrationOutcome {
+    Available(crate::RawTransactionIngress),
+    Missing(ksp_store_lib::RawTransactionReference),
+}
+
+#[cfg(test)]
+async fn hydrate_yellowstone_signal(
+    source: &crate::RawTransactionIngestYellowstoneSource,
+    settings: &crate::RawTransactionIngestSettings,
+    signal: RawTransactionIngestSourceSignal,
+    received_at: ksp_store_lib::RawTimestamp,
+) -> ksp_core_lib::Result<RawTransactionIngestHydrationOutcome> {
+    if &signal.network != settings.network() || signal.network != source.network {
+        return std::result::Result::Err(crate::runtime_error("hydration.network_mismatch"));
+    }
+    if signal.route != source.route {
+        return std::result::Result::Err(crate::runtime_error("hydration.source_route_mismatch"));
+    }
+    let commitment = match source.subscribe_request.commitment() {
+        std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) => ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => 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("hydration.commitment_invalid"));
+        },
+    };
+    let config = ksp_onchain_transport_lib::SolanaGetTransactionConfig::new(
+        std::option::Option::Some(commitment),
+        std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64),
+        std::option::Option::Some(0),
+    );
+    let signature_text = ksp_raw_transaction_lib::format_raw_transaction_signature(&signal.signature);
+    let observed = source.http_pool.get_transaction_observed(&source.hydration_role, signature_text.as_str(), std::option::Option::Some(&config)).await;
+    let observed = match observed {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(hydration_transport_error(error.code())),
+    };
+    let http_provider = observed.provider().as_str().to_owned();
+    let http_endpoint = observed.endpoint_name().to_owned();
+    let transaction = observed.into_value();
+    let transaction = match transaction {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => {
+            let reference = ksp_store_lib::RawTransactionReference::new(signal.network, signal.signature);
+            return std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Missing(reference));
+        },
+    };
+    if transaction.slot() != signal.slot {
+        return std::result::Result::Err(crate::runtime_error("hydration.slot_mismatch"));
+    }
+    if let std::option::Option::Some(expected_index) = signal.transaction_index {
+        if let ksp_onchain_transport_lib::SolanaWireField::Value(actual_index) = transaction.transaction_index() {
+            if expected_index != u64::from(*actual_index) {
+                return std::result::Result::Err(crate::runtime_error("hydration.transaction_index_mismatch"));
+            }
+        }
+    }
+    let transaction_data = match transaction.transaction() {
+        ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { data, encoding }
+            if *encoding == ksp_onchain_transport_lib::SolanaTransactionBinaryEncoding::Base64 =>
+        {
+            data.clone()
+        },
+        ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { .. }
+        | ksp_onchain_transport_lib::SolanaEncodedTransaction::LegacyBinary(_)
+        | ksp_onchain_transport_lib::SolanaEncodedTransaction::Json(_) => {
+            return std::result::Result::Err(crate::runtime_error("hydration.transaction_encoding_invalid"));
+        },
+    };
+    let embedded_signature = ksp_raw_transaction_lib::extract_raw_transaction_signature_from_binary_base64(transaction_data.as_str());
+    let embedded_signature = match embedded_signature {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("hydration.transaction_signature_invalid")),
+    };
+    if embedded_signature != signal.signature {
+        return std::result::Result::Err(crate::runtime_error("hydration.signature_mismatch"));
+    }
+    let provenance = build_hydration_provenance(settings, &signal, http_provider.as_str(), http_endpoint.as_str(), commitment, received_at);
+    let provenance = match provenance {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let source_key = hydration_source_key(&provenance);
+    let material = ksp_raw_transaction_lib::RawTransactionMaterial::binary_base64(
+        signal.network.clone(),
+        signal.signature,
+        transaction.slot(),
+        transaction.block_time(),
+        transaction_data,
+        map_hydration_wire_field(transaction.meta(), |value| return value.clone()),
+        map_hydration_wire_field(transaction.version(), |value| {
+            return match value {
+                ksp_onchain_transport_lib::SolanaTransactionVersion::Legacy => ksp_raw_transaction_lib::RawTransactionVersion::Legacy,
+                ksp_onchain_transport_lib::SolanaTransactionVersion::Number(number) => ksp_raw_transaction_lib::RawTransactionVersion::Number(*number),
+            };
+        }),
+        map_hydration_wire_field(transaction.transaction_index(), |value| return *value),
+    );
+    return std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Available(crate::RawTransactionIngress {
+        material,
+        network: signal.network,
+        provenance,
+        source_key,
+    }));
+}
+
+#[cfg(test)]
+fn build_hydration_provenance(
+    settings: &crate::RawTransactionIngestSettings,
+    signal: &RawTransactionIngestSourceSignal,
+    http_provider: &str,
+    http_endpoint: &str,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    received_at: ksp_store_lib::RawTimestamp,
+) -> ksp_core_lib::Result<ksp_store_lib::RawAcquisitionProvenance> {
+    let (provider, endpoint_id) = match composite_provenance_codes(&signal.route, http_provider, http_endpoint) {
+        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("hydration.protocol_unrepresentable")),
+    };
+    let acquisition_method = match ksp_store_lib::RawProvenanceCode::new(hydration_method_code(signal.family)) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("hydration.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("hydration.capture_session_unrepresentable")),
+    };
+    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("hydration.commitment_unrepresentable")),
+    };
+    let filter_id = match signal.matched_filter_id.clone() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => match fingerprint_filter_code(&signal.matched_filter_fingerprint) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+    };
+    let mut provenance =
+        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)
+            .with_filter_id(filter_id);
+    if let std::option::Option::Some(observed_at) = representable_observed_at(signal.created_at, received_at) {
+        provenance = match provenance.try_with_observed_at(observed_at) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("hydration.observed_at_invalid")),
+        };
+    }
+    return std::result::Result::Ok(provenance);
+}
+
+#[cfg(test)]
+fn hydration_method_code(family: RawTransactionIngestSourceFamily) -> &'static str {
+    return match family {
+        RawTransactionIngestSourceFamily::Transaction => "transaction_get_transaction",
+        RawTransactionIngestSourceFamily::TransactionStatus => "status_get_transaction",
+    };
+}
+
+#[cfg(test)]
+fn fingerprint_filter_code(fingerprint: &[u8; 32]) -> ksp_core_lib::Result<ksp_store_lib::RawProvenanceCode> {
+    const HEX: &[u8; 16] = b"0123456789abcdef";
+    let mut value = std::string::String::with_capacity(71);
+    value.push_str("sha256.");
+    for byte in fingerprint {
+        value.push(char::from(HEX[usize::from(*byte >> 4)]));
+        value.push(char::from(HEX[usize::from(*byte & 0x0f)]));
+    }
+    return match ksp_store_lib::RawProvenanceCode::new(value) {
+        std::result::Result::Ok(value) => std::result::Result::Ok(value),
+        std::result::Result::Err(_) => std::result::Result::Err(crate::runtime_error("hydration.filter_fingerprint_unrepresentable")),
+    };
+}
+
+#[cfg(test)]
+fn representable_observed_at(
+    created_at: std::option::Option<RawTransactionIngestSourceTimestamp>,
+    received_at: ksp_store_lib::RawTimestamp,
+) -> std::option::Option<ksp_store_lib::RawTimestamp> {
+    let created_at = match created_at {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    let seconds = match u64::try_from(created_at.seconds) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    let millis = match seconds.checked_mul(1_000).and_then(|value| return value.checked_add(u64::from(created_at.nanos / 1_000_000))) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    let observed_at = match ksp_store_lib::RawTimestamp::from_unix_millis(millis) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    if observed_at > received_at {
+        return std::option::Option::None;
+    }
+    return std::option::Option::Some(observed_at);
+}
+
+#[cfg(test)]
+fn hydration_source_key(provenance: &ksp_store_lib::RawAcquisitionProvenance) -> [u8; 32] {
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN);
+    hash_hydration_source_key(&mut hasher, provenance.provider().as_str().as_bytes());
+    hash_hydration_source_key(&mut hasher, provenance.protocol().as_str().as_bytes());
+    hash_hydration_source_key(&mut hasher, provenance.acquisition_method().as_str().as_bytes());
+    if let std::option::Option::Some(value) = provenance.endpoint_id() {
+        hash_hydration_source_key(&mut hasher, value.as_str().as_bytes());
+    }
+    if let std::option::Option::Some(value) = provenance.commitment() {
+        hash_hydration_source_key(&mut hasher, value.as_str().as_bytes());
+    }
+    if let std::option::Option::Some(value) = provenance.capture_session_id() {
+        hash_hydration_source_key(&mut hasher, value.as_str().as_bytes());
+    }
+    if let std::option::Option::Some(value) = provenance.filter_id() {
+        hash_hydration_source_key(&mut hasher, value.as_str().as_bytes());
+    }
+    return hasher.finalize().into();
+}
+
+#[cfg(test)]
+fn hash_hydration_source_key(hasher: &mut sha2::Sha256, value: &[u8]) {
+    hasher.update((value.len() as u64).to_be_bytes());
+    hasher.update(value);
+    return;
+}
+
+#[cfg(test)]
+fn hydration_transport_error(code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED, "RAW transaction ingest hydration transport failed")
+        .with_context("transport_domain", code.domain())
+        .with_context("transport_code", code.code());
+}
+
+#[cfg(test)]
+fn map_hydration_wire_field<T, U, F>(
+    field: &ksp_onchain_transport_lib::SolanaWireField<T>,
+    mut map_value: F,
+) -> ksp_raw_transaction_lib::RawTransactionWireField<U>
+where
+    F: FnMut(&T) -> U,
+{
+    return match field {
+        ksp_onchain_transport_lib::SolanaWireField::Omitted => ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
+        ksp_onchain_transport_lib::SolanaWireField::Null => ksp_raw_transaction_lib::RawTransactionWireField::Null,
+        ksp_onchain_transport_lib::SolanaWireField::Value(value) => ksp_raw_transaction_lib::RawTransactionWireField::Value(map_value(value)),
+    };
+}
+
 #[cfg(test)]
 #[path = "../unit_tests/runtime_resources.rs"]
 mod tests;