refactor(iroh): Remove unused rate limiter (#3007)

## Description

This was a rate limiter which seems like the intention was to slow
down how fast it sends to a particular relay server.  This is the
wrong place to rate-limit.  If the server limits that it should read
slower from the TCP stream (it does this now!) and that blocks up the
TCP-stream and in turn slows down the relay client.  In no world does
a socket internally rate-limit to one of it's destinations, that's not
how sockets work.  And this client exists for the MagicSock.

## 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

- [x] Self-review.
- [x] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [x] Tests if relevant.
- [x] All breaking changes documented.

Author: flub

iroh-relay/src/client/conn.rs

@@ -4,14 +4,13 @@
 
 use std::{
     net::SocketAddr,
-    num::NonZeroU32,
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
     time::Duration,
 };
 
-use anyhow::{anyhow, bail, ensure, Context as _, Result};
+use anyhow::{anyhow, bail, ensure, Result};
 use bytes::Bytes;
 use futures_lite::Stream;
 use futures_sink::Sink;
@@ -229,15 +228,14 @@ enum ConnWriterMessage {
 struct ConnWriterTasks {
     recv_msgs: mpsc::Receiver<ConnWriterMessage>,
     writer: ConnWriter,
-    rate_limiter: Option<RateLimiter>,
 }
 
 impl ConnWriterTasks {
     async fn run(mut self) -> Result<()> {
         while let Some(msg) = self.recv_msgs.recv().await {
             match msg {
                 ConnWriterMessage::Packet((key, bytes)) => {
-                    send_packet(&mut self.writer, &self.rate_limiter, key, bytes).await?;
+                    send_packet(&mut self.writer, key, bytes).await?;
                 }
                 ConnWriterMessage::Pong(data) => {
                     write_frame(&mut self.writer, Frame::Pong { data }, None).await?;
@@ -360,7 +358,7 @@ impl ConnBuilder {
         }
     }
 
-    async fn server_handshake(&mut self) -> Result<Option<RateLimiter>> {
+    async fn server_handshake(&mut self) -> Result<()> {
         debug!("server_handshake: started");
         let client_info = ClientInfo {
             version: PROTOCOL_VERSION,
@@ -369,22 +367,18 @@ impl ConnBuilder {
         crate::protos::relay::send_client_key(&mut self.writer, &self.secret_key, &client_info)
             .await?;
 
-        // TODO: add some actual configuration
-        let rate_limiter = RateLimiter::new(0, 0)?;
-
         debug!("server_handshake: done");
-        Ok(rate_limiter)
+        Ok(())
     }
 
     pub async fn build(mut self) -> Result<(Conn, ConnReceiver)> {
         // exchange information with the server
-        let rate_limiter = self.server_handshake().await?;
+        self.server_handshake().await?;
 
         // create task to handle writing to the server
         let (writer_sender, writer_recv) = mpsc::channel(PER_CLIENT_SEND_QUEUE_DEPTH);
         let writer_task = tokio::task::spawn(
             ConnWriterTasks {
-                rate_limiter,
                 writer: self.writer,
                 recv_msgs: writer_recv,
             }
@@ -494,7 +488,6 @@ pub enum ReceivedMessage {
 
 pub(crate) async fn send_packet<S: Sink<Frame, Error = std::io::Error> + Unpin>(
     mut writer: S,
-    rate_limiter: &Option<RateLimiter>,
     dst: NodeId,
     packet: Bytes,
 ) -> Result<()> {
@@ -508,43 +501,8 @@ pub(crate) async fn send_packet<S: Sink<Frame, Error = std::io::Error> + Unpin>(
         dst_key: dst,
         packet,
     };
-    if let Some(rate_limiter) = rate_limiter {
-        if rate_limiter.check_n(frame.len()).is_err() {
-            tracing::debug!("dropping send: rate limit reached");
-            return Ok(());
-        }
-    }
     writer.send(frame).await?;
     writer.flush().await?;
 
     Ok(())
 }
-
-pub(crate) struct RateLimiter {
-    inner: governor::DefaultDirectRateLimiter,
-}
-
-impl RateLimiter {
-    pub(crate) fn new(bytes_per_second: usize, bytes_burst: usize) -> Result<Option<Self>> {
-        if bytes_per_second == 0 || bytes_burst == 0 {
-            return Ok(None);
-        }
-        let bytes_per_second = NonZeroU32::new(u32::try_from(bytes_per_second)?)
-            .context("bytes_per_second not non-zero")?;
-        let bytes_burst =
-            NonZeroU32::new(u32::try_from(bytes_burst)?).context("bytes_burst not non-zero")?;
-        Ok(Some(Self {
-            inner: governor::RateLimiter::direct(
-                governor::Quota::per_second(bytes_per_second).allow_burst(bytes_burst),
-            ),
-        }))
-    }
-
-    pub(crate) fn check_n(&self, n: usize) -> Result<()> {
-        let n = NonZeroU32::new(u32::try_from(n)?).context("n not non-zero")?;
-        match self.inner.check_n(n) {
-            Ok(_) => Ok(()),
-            Err(_) => bail!("batch cannot go through"),
-        }
-    }
-}

iroh-relay/src/server/actor.rs

@@ -308,8 +308,7 @@ mod tests {
 
         // write message from b to a
         let msg = b"hello world!";
-        crate::client::conn::send_packet(&mut b_io, &None, node_id_a, Bytes::from_static(msg))
-            .await?;
+        crate::client::conn::send_packet(&mut b_io, node_id_a, Bytes::from_static(msg)).await?;
 
         // get message on a's reader
         let frame = recv_frame(FrameType::RecvPacket, &mut a_io).await?;

iroh-relay/src/server/client_conn.rs

@@ -617,7 +617,7 @@ mod tests {
         // send packet
         println!("  send packet");
         let data = b"hello world!";
-        conn::send_packet(&mut io_rw, &None, target, Bytes::from_static(data)).await?;
+        conn::send_packet(&mut io_rw, target, Bytes::from_static(data)).await?;
         let msg = server_channel_r.recv().await.unwrap();
         match msg {
             actor::Message::SendPacket {
@@ -640,7 +640,7 @@ mod tests {
         let mut disco_data = disco::MAGIC.as_bytes().to_vec();
         disco_data.extend_from_slice(target.as_bytes());
         disco_data.extend_from_slice(data);
-        conn::send_packet(&mut io_rw, &None, target, disco_data.clone().into()).await?;
+        conn::send_packet(&mut io_rw, target, disco_data.clone().into()).await?;
         let msg = server_channel_r.recv().await.unwrap();
         match msg {
             actor::Message::SendDiscoPacket {
@@ -698,7 +698,7 @@ mod tests {
         let data = b"hello world!";
         let target = SecretKey::generate().public();
 
-        conn::send_packet(&mut io_rw, &None, target, Bytes::from_static(data)).await?;
+        conn::send_packet(&mut io_rw, target, Bytes::from_static(data)).await?;
         let msg = server_channel_r.recv().await.unwrap();
         match msg {
             actor::Message::SendPacket {