Remove blocking async overlay events loop

Author: njgheorghita

newsfragments/382.fixed.md

@@ -0,0 +1 @@
+Remove blocking async from overlay events loop.

trin-history/src/events.rs

@@ -16,7 +16,7 @@ impl HistoryEvents {
         loop {
             tokio::select! {
                 Some(talk_request) = self.event_rx.recv() => {
-                    self.handle_history_talk_request(talk_request).await;
+                    self.handle_history_talk_request(talk_request);
                 }
                 Some(event) = self.utp_listener_rx.recv() => {
                     if let Err(err) = self.network.overlay.process_utp_event(event) {
@@ -28,25 +28,27 @@ impl HistoryEvents {
     }
 
     /// Handle history network TalkRequest event
-    async fn handle_history_talk_request(&self, talk_request: TalkRequest) {
-        let reply = match self
-            .network
-            .overlay
-            .process_one_request(&talk_request)
-            .instrument(tracing::info_span!("history_network"))
-            .await
-        {
-            Ok(response) => {
-                debug!("Sending reply: {:?}", response);
-                Message::from(response).into()
-            }
-            Err(error) => {
-                error!("Failed to process portal history event: {error}");
-                error.to_string().into_bytes()
+    fn handle_history_talk_request(&self, talk_request: TalkRequest) {
+        let network = Arc::clone(&self.network);
+        tokio::spawn(async move {
+            let reply = match network
+                .overlay
+                .process_one_request(&talk_request)
+                .instrument(tracing::info_span!("history_network"))
+                .await
+            {
+                Ok(response) => {
+                    debug!("Sending reply: {:?}", response);
+                    Message::from(response).into()
+                }
+                Err(error) => {
+                    error!("Failed to process portal history event: {error}");
+                    error.to_string().into_bytes()
+                }
+            };
+            if let Err(error) = talk_request.respond(reply) {
+                warn!("Failed to send reply: {error}");
             }
-        };
-        if let Err(error) = talk_request.respond(reply) {
-            warn!("Failed to send reply: {error}");
-        }
+        });
     }
 }

trin-state/src/events.rs

@@ -16,7 +16,7 @@ impl StateEvents {
         loop {
             tokio::select! {
                 Some(talk_request) = self.event_rx.recv() => {
-                    self.handle_state_talk_request(talk_request).await;
+                    self.handle_state_talk_request(talk_request);
                 }
                 Some(event) = self.utp_listener_rx.recv() => {
                     if let Err(err) = self.network.overlay.process_utp_event(event) {
@@ -28,25 +28,27 @@ impl StateEvents {
     }
 
     /// Handle state network TalkRequest event
-    async fn handle_state_talk_request(&self, talk_request: TalkRequest) {
-        let reply = match self
-            .network
-            .overlay
-            .process_one_request(&talk_request)
-            .instrument(tracing::info_span!("state_network"))
-            .await
-        {
-            Ok(response) => {
-                debug!("Sending reply: {:?}", response);
-                Message::from(response).into()
-            }
-            Err(error) => {
-                error!("Failed to process portal state event: {error}");
-                error.to_string().into_bytes()
+    fn handle_state_talk_request(&self, talk_request: TalkRequest) {
+        let network = Arc::clone(&self.network);
+        tokio::spawn(async move {
+            let reply = match network
+                .overlay
+                .process_one_request(&talk_request)
+                .instrument(tracing::info_span!("state_network"))
+                .await
+            {
+                Ok(response) => {
+                    debug!("Sending reply: {:?}", response);
+                    Message::from(response).into()
+                }
+                Err(error) => {
+                    error!("Failed to process portal state event: {error}");
+                    error.to_string().into_bytes()
+                }
+            };
+            if let Err(error) = talk_request.respond(reply) {
+                warn!("Failed to send reply: {error}");
             }
-        };
-        if let Err(error) = talk_request.respond(reply) {
-            warn!("Failed to send reply: {error}");
-        }
+        });
     }
 }