v0.3.15-pre.009-fix.001

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: 45
+// version: 46
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -401,6 +401,16 @@ impl RawTransactionIngestLiveSource {
         );
     }
 
+    fn kind_code(&self) -> &'static str {
+        return match self {
+            Self::HeliusTransaction(_) => "helius_transaction",
+            Self::HttpBlockPolling(_) => "http_block_polling",
+            Self::StandardBlock(_) => "standard_block",
+            Self::StandardLogs(_) => "standard_logs",
+            Self::Yellowstone(_) => "yellowstone",
+        };
+    }
+
     fn source_key(&self) -> [u8; 32] {
         return match self {
             Self::HeliusTransaction(source) => source.source_key,
@@ -427,6 +437,7 @@ impl RawTransactionIngestLiveSource {
         inventory_publisher: RawTransactionIngestSourceInventoryPublisher,
         shared: RawTransactionIngestSourceRuntimeShared,
     ) -> ksp_core_lib::Result<()> {
+        let source_kind = self.kind_code();
         let (source_frontier_sender, mut source_frontier_receiver) =
             tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
         let mut source_future = std::boxed::Box::pin(async move {
@@ -453,6 +464,20 @@ impl RawTransactionIngestLiveSource {
                     if let std::result::Result::Err(error) = inventory_publisher.publish(source_projection_with_state(latest, terminal_state)) {
                         return std::result::Result::Err(error);
                     }
+                    if let std::result::Result::Err(error) = &result {
+                        let transport_domain = source_error_context_value(error, "transport_domain").unwrap_or("none");
+                        let transport_code = source_error_context_value(error, "transport_code").unwrap_or("none");
+                        ksp_logging_lib::warn!(
+                            target: "ksp-worker-raw-transaction-ingest-lib",
+                            domain = "raw_transaction_ingest.source",
+                            source_kind = source_kind,
+                            error_domain = error.code().domain(),
+                            error_code = error.code().code(),
+                            transport_domain = transport_domain,
+                            transport_code = transport_code,
+                            "RAW transaction ingest live source reached terminal failure"
+                        );
+                    }
                     return result;
                 }
                 changed = source_frontier_receiver.changed() => {
@@ -1224,14 +1249,14 @@ impl crate::RawTransactionIngestYellowstoneSource {
             processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
             return std::result::Result::Err(error);
         }
-        return match closed {
+        return match resolve_yellowstone_close_after_stop(*stop_receiver.borrow(), 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()))
+                std::result::Result::Err(error)
             },
         };
     }
@@ -1337,14 +1362,14 @@ impl crate::RawTransactionIngestYellowstoneSource {
             processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
             return std::result::Result::Err(error);
         }
-        return match closed {
+        return match resolve_yellowstone_close_after_stop(*stop_receiver.borrow(), 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()))
+                std::result::Result::Err(error)
             },
         };
     }
@@ -4338,6 +4363,23 @@ fn source_transport_error(code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Error
         .with_context("transport_code", code.code());
 }
 
+fn source_error_context_value<'a>(error: &'a ksp_core_lib::Error, key: &'static str) -> std::option::Option<&'a str> {
+    for context in error.context() {
+        if context.key() == key {
+            return std::option::Option::Some(context.value());
+        }
+    }
+    return std::option::Option::None;
+}
+
+fn resolve_yellowstone_close_after_stop(stop_requested: bool, closed: ksp_core_lib::Result<()>) -> ksp_core_lib::Result<()> {
+    return match closed {
+        std::result::Result::Ok(()) => std::result::Result::Ok(()),
+        std::result::Result::Err(error) if stop_requested && error.code() == ksp_onchain_transport_lib::ERROR_CODE_TIMEOUT => std::result::Result::Ok(()),
+        std::result::Result::Err(error) => std::result::Result::Err(source_transport_error(error.code())),
+    };
+}
+
 fn map_yellowstone_source_state(state: ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState) -> crate::RawTransactionIngestSourceState {
     return match state {
         ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Active => crate::RawTransactionIngestSourceState::Active,