fix(iroh-relay): don't stop relay client actor if queues become full (#3294)

## Description

This is a fix to the iroh-relay server.

Current situation: If a client stream becomes full (because the client
is not reading fast enough from the network), the internal queue where
messages to that client are queued becomes full too. Now, when another
client wants to send a message to the filled-up client, instead of
skipping over the message we propagate the error such that the sending
client's actor dies. This is bad, we should instead just drop the
message.

This PR changes the behavior to log the failed forward (at WARN level)
but does not stop the sending client's actor.

It also adds a test that fails without the change and passes with the
change.

## Breaking Changes

<!-- Optional, if there are any breaking changes document them,
including how to migrate older code. -->

## Notes & open questions

<!-- Any notes, remarks or open questions you have to make about the PR.
-->

## Change checklist
<!-- Remove any that are not relevant. -->
- [x] Self-review.
- [x] Tests if relevant.

Author: Frando

iroh-relay/src/server.rs

@@ -1252,4 +1252,37 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    #[traced_test]
+    async fn test_relay_clients_full() -> TestResult<()> {
+        let server = spawn_local_relay().await.unwrap();
+        let relay_url = format!("http://{}", server.http_addr().unwrap());
+        let relay_url: RelayUrl = relay_url.parse().unwrap();
+
+        // set up client a
+        let a_secret_key = SecretKey::generate(rand::thread_rng());
+        let resolver = dns_resolver();
+        let mut client_a = ClientBuilder::new(relay_url.clone(), a_secret_key, resolver.clone())
+            .connect()
+            .await?;
+
+        // set up client b
+        let b_secret_key = SecretKey::generate(rand::thread_rng());
+        let b_key = b_secret_key.public();
+        let _client_b = ClientBuilder::new(relay_url.clone(), b_secret_key, resolver.clone())
+            .connect()
+            .await?;
+
+        // send messages from a to b, without b receiving anything.
+        // we should still keep succeeding to send, even if the packet won't be forwarded
+        // by the relay server because the server's send queue for b fills up.
+        let msg = Bytes::from("hello, b");
+        for _i in 0..1000 {
+            client_a
+                .send(SendMessage::SendPacket(b_key, msg.clone()))
+                .await?;
+        }
+        Ok(())
+    }
 }

iroh-relay/src/server/client.rs

@@ -365,7 +365,11 @@ impl Actor {
         match frame {
             Frame::SendPacket { dst_key, packet } => {
                 let packet_len = packet.len();
-                self.handle_frame_send_packet(dst_key, packet)?;
+                if let Err(err @ ForwardPacketError { .. }) =
+                    self.handle_frame_send_packet(dst_key, packet)
+                {
+                    warn!("failed to handle send packet frame: {err:#}");
+                }
                 self.metrics.bytes_recv.inc_by(packet_len as u64);
             }
             Frame::Ping { data } => {
@@ -387,7 +391,7 @@ impl Actor {
         Ok(())
     }
 
-    fn handle_frame_send_packet(&self, dst: NodeId, data: Bytes) -> Result<()> {
+    fn handle_frame_send_packet(&self, dst: NodeId, data: Bytes) -> Result<(), ForwardPacketError> {
         if disco::looks_like_disco_wrapper(&data) {
             self.metrics.disco_packets_recv.inc();
             self.clients
@@ -401,6 +405,31 @@ impl Actor {
     }
 }
 
+#[derive(Debug)]
+pub(crate) enum PacketScope {
+    Disco,
+    Data,
+}
+
+#[derive(Debug)]
+pub(crate) enum SendError {
+    Full,
+    Closed,
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error("failed to forward {scope:?} packet: {reason:?}")]
+pub(crate) struct ForwardPacketError {
+    scope: PacketScope,
+    reason: SendError,
+}
+
+impl ForwardPacketError {
+    pub(crate) fn new(scope: PacketScope, reason: SendError) -> Self {
+        Self { scope, reason }
+    }
+}
+
 /// Rate limiter for reading from a [`RelayedStream`].
 ///
 /// The writes to the sink are not rate limited.

iroh-relay/src/server/clients.rs

@@ -9,15 +9,18 @@ use std::{
     },
 };
 
-use anyhow::{bail, Result};
+use anyhow::Result;
 use bytes::Bytes;
 use dashmap::DashMap;
 use iroh_base::NodeId;
 use tokio::sync::mpsc::error::TrySendError;
 use tracing::{debug, trace};
 
-use super::client::{Client, Config};
-use crate::server::metrics::Metrics;
+use super::client::{Client, Config, ForwardPacketError};
+use crate::server::{
+    client::{PacketScope, SendError},
+    metrics::Metrics,
+};
 
 /// Manages the connections to all currently connected clients.
 #[derive(Debug, Default, Clone)]
@@ -109,7 +112,7 @@ impl Clients {
         data: Bytes,
         src: NodeId,
         metrics: &Metrics,
-    ) -> Result<()> {
+    ) -> Result<(), ForwardPacketError> {
         let Some(client) = self.0.clients.get(&dst) else {
             debug!(dst = dst.fmt_short(), "no connected client, dropped packet");
             metrics.send_packets_dropped.inc();
@@ -126,15 +129,18 @@ impl Clients {
                     dst = dst.fmt_short(),
                     "client too busy to receive packet, dropping packet"
                 );
-                bail!("failed to send message: full");
+                Err(ForwardPacketError::new(PacketScope::Data, SendError::Full))
             }
             Err(TrySendError::Closed(_)) => {
                 debug!(
                     dst = dst.fmt_short(),
                     "can no longer write to client, dropping message and pruning connection"
                 );
                 client.start_shutdown();
-                bail!("failed to send message: gone");
+                Err(ForwardPacketError::new(
+                    PacketScope::Data,
+                    SendError::Closed,
+                ))
             }
         }
     }
@@ -146,7 +152,7 @@ impl Clients {
         data: Bytes,
         src: NodeId,
         metrics: &Metrics,
-    ) -> Result<()> {
+    ) -> Result<(), ForwardPacketError> {
         let Some(client) = self.0.clients.get(&dst) else {
             debug!(
                 dst = dst.fmt_short(),
@@ -166,15 +172,18 @@ impl Clients {
                     dst = dst.fmt_short(),
                     "client too busy to receive disco packet, dropping packet"
                 );
-                bail!("failed to send message: full");
+                Err(ForwardPacketError::new(PacketScope::Disco, SendError::Full))
             }
             Err(TrySendError::Closed(_)) => {
                 debug!(
                     dst = dst.fmt_short(),
                     "can no longer write to client, dropping disco message and pruning connection"
                 );
                 client.start_shutdown();
-                bail!("failed to send message: gone");
+                Err(ForwardPacketError::new(
+                    PacketScope::Disco,
+                    SendError::Closed,
+                ))
             }
         }
     }