v0.3.14-pre.007

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: 32
+// version: 33
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -4026,6 +4026,20 @@ struct RawTransactionIngestHydrationFetch {
     observed: RawTransactionIngestObservedTransaction,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RawTransactionIngestKnownReferenceMissingDisposition {
+    AwaitCoverage,
+    PreferBlockSlot,
+}
+
+enum RawTransactionIngestKnownReferenceHydrationResolution {
+    Available(crate::RawTransactionIngress),
+    Missing {
+        disposition: RawTransactionIngestKnownReferenceMissingDisposition,
+        obligation: crate::RawTransactionIngestKnownReferenceObligation,
+    },
+}
+
 #[derive(Clone)]
 enum RawTransactionIngestSharedHydrationResult {
     Available(RawTransactionIngestObservedTransaction),
@@ -4219,14 +4233,24 @@ impl RawTransactionIngestHydrationCoordinator {
         self.pending_signal_count -= pending.signals.len();
         for pending_signal in pending.signals {
             let signal_slot = pending_signal.signal.slot;
-            let ingress = finalize_hydration(hydration, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed);
-            let ingress = match ingress {
+            let resolution = resolve_known_reference_hydration(hydration, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed);
+            let resolution = match resolution {
                 std::result::Result::Ok(value) => value,
                 std::result::Result::Err(error) => return std::result::Result::Err(error),
             };
-            let ingress = match ingress {
-                std::option::Option::Some(value) => value,
-                std::option::Option::None => {
+            let ingress = match resolution {
+                RawTransactionIngestKnownReferenceHydrationResolution::Available(value) => value,
+                RawTransactionIngestKnownReferenceHydrationResolution::Missing { disposition, obligation } => {
+                    if obligation.commitment() != hydration.commitment
+                        || obligation.reference().network() != &hydration.network
+                        || obligation.slot() != signal_slot
+                    {
+                        return std::result::Result::Err(crate::runtime_error("source.known_reference_obligation_mismatch"));
+                    }
+                    match disposition {
+                        RawTransactionIngestKnownReferenceMissingDisposition::AwaitCoverage
+                        | RawTransactionIngestKnownReferenceMissingDisposition::PreferBlockSlot => {},
+                    }
                     if let std::result::Result::Err(error) = processing_frontier.settle_pending(signal_slot) {
                         return std::result::Result::Err(error);
                     }
@@ -4389,6 +4413,38 @@ fn shared_hydration_result(
     };
 }
 
+fn resolve_known_reference_hydration(
+    hydration: &RawTransactionIngestHydrationContext,
+    settings: &crate::RawTransactionIngestSettings,
+    signal: RawTransactionIngestSourceSignal,
+    received_at: ksp_store_lib::RawTimestamp,
+    observed: &RawTransactionIngestObservedTransaction,
+) -> ksp_core_lib::Result<RawTransactionIngestKnownReferenceHydrationResolution> {
+    let reference = ksp_store_lib::RawTransactionReference::new(signal.network.clone(), signal.signature);
+    let obligation = match crate::RawTransactionIngestKnownReferenceObligation::new(reference, signal.slot, hydration.commitment) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let ingress = finalize_hydration(hydration, settings, signal, received_at, observed);
+    let ingress = match ingress {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if let std::option::Option::Some(value) = ingress {
+        return std::result::Result::Ok(RawTransactionIngestKnownReferenceHydrationResolution::Available(value));
+    }
+    let block_slot_supported = match http_role_supports_rpc_method(&hydration.http_pool, &hydration.hydration_role, "getBlock", hydration.network.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let disposition = if block_slot_supported {
+        RawTransactionIngestKnownReferenceMissingDisposition::PreferBlockSlot
+    } else {
+        RawTransactionIngestKnownReferenceMissingDisposition::AwaitCoverage
+    };
+    return std::result::Result::Ok(RawTransactionIngestKnownReferenceHydrationResolution::Missing { disposition, obligation });
+}
+
 fn finalize_hydration(
     hydration: &RawTransactionIngestHydrationContext,
     settings: &crate::RawTransactionIngestSettings,