Fix issues with streams stalling (#65)

* add option for tracing

* bump max request/response size

* use while let

* fix unyielding streams

* introduce max_rpc_payload_size

---------

Co-authored-by: Seun Lanlege <seun@polytope.technology>

client/src/any.rs

@@ -279,7 +279,7 @@ macro_rules! chain {
             async fn state_machine_update_notification(
                 &self,
                 counterparty_state_id: ismp::consensus::StateMachineId,
-            ) -> Result<primitives::BoxStream<primitives::StateMachineUpdated>, anyhow::Error> {
+            ) -> Result<primitives::BoxStream<primitives::StreamItem>, anyhow::Error> {
                 match self {
 					$(
 						$(#[$($meta)*])*

consensus/src/lib.rs

@@ -29,24 +29,30 @@ where
 	let task_a = {
 		let chain_a = chain_a.clone();
 		let chain_b = chain_b.clone();
-		Box::pin(handle_notification(chain_a, chain_b))
+		tokio::spawn(async move {
+			let _ = handle_notification(chain_a, chain_b).await?;
+			Ok::<_, anyhow::Error>(())
+		})
 	};
 
 	let task_b = {
 		let chain_a = chain_a.clone();
 		let chain_b = chain_b.clone();
-		Box::pin(handle_notification(chain_b, chain_a))
+		tokio::spawn(async move {
+			let _ = handle_notification(chain_b, chain_a).await?;
+			Ok::<_, anyhow::Error>(())
+		})
 	};
 
 	// if one task completes, abort the other
 	tokio::select! {
 		result_a = task_a => {
-			result_a?
+			result_a??
 		}
 		result_b = task_b => {
-			result_b?
+			result_b??
 		}
-	};
+	}
 
 	Ok(())
 }