Close tainted events subscription if node returns un-deserializable d…

…ata.

api/bin/chainflip-broker-api/src/main.rs

@@ -17,9 +17,9 @@ use chainflip_api::{
 };
 use clap::Parser;
 use custom_rpc::CustomApiClient;
-use futures::{FutureExt, TryStreamExt};
+use futures::{FutureExt, StreamExt};
 use jsonrpsee::{
-	core::{async_trait, ClientError},
+	core::{async_trait, ClientError, SubscriptionResult},
 	proc_macros::rpc,
 	server::ServerBuilder,
 	types::{ErrorCode, ErrorObject, ErrorObjectOwned},
@@ -30,7 +30,7 @@ use std::{
 	path::PathBuf,
 	sync::{atomic::AtomicBool, Arc},
 };
-use tracing::{event, log, Level};
+use tracing::log;
 
 #[derive(thiserror::Error, Debug)]
 pub enum BrokerApiError {
@@ -124,7 +124,7 @@ pub trait Rpc {
 	) -> RpcResult<ChainAccounts>;
 
 	#[subscription(name = "subscribe_tainted_transaction_events", item = Result<BlockUpdate<TaintedTransactionEvents>,String>)]
-	async fn subscribe_tainted_transaction_events(&self);
+	async fn subscribe_tainted_transaction_events(&self) -> SubscriptionResult;
 }
 
 pub struct RpcServerImpl {
@@ -247,29 +247,30 @@ impl RpcServer for RpcServerImpl {
 			.map_err(BrokerApiError::ClientError)
 	}
 
-	async fn subscribe_tainted_transaction_events(&self, pending_sink: PendingSubscriptionSink) {
-		let result = self
+	async fn subscribe_tainted_transaction_events(&self, pending_sink: PendingSubscriptionSink) -> SubscriptionResult {
+		let subscription_stream = self
 			.api
 			.state_chain_client
 			.base_rpc_client
 			.raw_rpc_client
 			.cf_subscribe_tainted_transaction_events()
-			.await;
+			.await?
+			.take_while(|item| {
+				futures::future::ready(
+					item.as_ref()
+						.inspect_err(|err| {
+							tracing::error!("Subscription error, could not deserialize message: {err}. Closing subscription.")
+						})
+						.is_ok(),
+				)
+			})
+			.map(Result::unwrap);
 
-		match result {
-			Ok(subscription) => {
-				tokio::spawn(async {
-					sc_rpc::utils::pipe_from_stream(
-						pending_sink,
-						subscription.map_err(|err| format!("{err}")),
-					)
-					.await
-				});
-			},
-			Err(err) => {
-				event!(Level::ERROR, "Could not subscribe to `tainted_btc_transaction_events` subscription on the node. Error: {err}");
-			},
-		}
+		tokio::spawn(async {
+			sc_rpc::utils::pipe_from_stream(pending_sink, subscription_stream).await
+		});
+
+		Ok(())
 	}
 }