refactor: apply backpressure if relay disco recv queue fills up

Author: Frando

iroh/src/magicsock/relay_actor.rs

@@ -618,7 +618,7 @@ impl ActiveRelayActor {
                     };
                     match msg {
                         Ok(msg) => {
-                            self.handle_relay_msg(msg, &mut state);
+                            self.handle_relay_msg(msg, &mut state).await;
                             // reset the ping timer, we have just received a message
                             ping_interval.reset();
                         },
@@ -641,7 +641,7 @@ impl ActiveRelayActor {
         res.map_err(|err| state.map_err(err))
     }
 
-    fn handle_relay_msg(&mut self, msg: ReceivedMessage, state: &mut ConnectedRelayState) {
+    async fn handle_relay_msg(&mut self, msg: ReceivedMessage, state: &mut ConnectedRelayState) {
         match msg {
             ReceivedMessage::ReceivedPacket {
                 remote_node_id,
@@ -676,7 +676,7 @@ impl ActiveRelayActor {
                                 relay_url: datagram.url.clone(),
                                 relay_remote_node_id: datagram.src,
                             };
-                            if let Err(err) = self.relay_disco_recv.try_send(message) {
+                            if let Err(err) = self.relay_disco_recv.send(message).await {
                                 warn!("Dropping received relay disco packet: {err:#}");
                             }
                         }
@@ -774,7 +774,7 @@ impl ActiveRelayActor {
                         break Err(anyhow!("Stream closed by server."));
                     };
                     match msg {
-                        Ok(msg) => self.handle_relay_msg(msg, state),
+                        Ok(msg) => self.handle_relay_msg(msg, state).await,
                         Err(err) => break Err(anyhow!("Client stream read error: {err:#}")),
                     }
                 }