refactor(iroh,iroh-relay): Remove legacy relay path, make websocket connections default (#3384)

## Description

Split out from #3331. 

Switch to websockets and remove the old code path.

Also:
- I've removed `governor` for rate limiting
- We used to guess the amount of bytes a frame was encoded as after
we've decoded it, because that's how governor works (you do it on
`Stream`s, not the underlying `AsyncRead`).
- Instead we have our own couple of lines of rate-limiting logic (with
tests) on the `AsyncRead` level.
- The reason is we can't guess the size of websocket frames as easily
anymore.

## Breaking Changes

In iroh:
- Removed `iroh::endpoint::Builder::relay_conn_protocol`. It's now
always websockets
- Removed the `iroh::RelayProtocol` re-export (the type was removed in
`iroh-relay`)

In iroh-relay:
- Removed `ClientBuilder::protocol`
- Removed `http::Protocol` type
- Renamed `frames_rx_ratelimited_total` metric to
`bytes_rx_ratelimited_total`, which now tracks bytes not frames
- Removed `relay_accepts` and `websocket_accepts` metrics

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

Author: matheus23

Cargo.lock

@@ -89,7 +89,7 @@ rustls-cert-file-reader = { version = "0.4.1", optional = true }
 rustls-pemfile = { version = "2.1", optional = true }
 time = { version = "0.3.37", optional = true }
 tokio-rustls-acme = { version = "0.7.1", optional = true }
-tokio-websockets = { version = "0.11.3", features = ["rustls-bring-your-own-connector", "ring", "getrandom", "rand", "server"], optional = true } # server-side websocket implementation
+tokio-websockets = { version = "0.12", features = ["rustls-bring-your-own-connector", "ring", "getrandom", "rand", "server"], optional = true } # server-side websocket implementation
 simdutf8 = { version = "0.1.5", optional = true } # minimal version fix
 toml = { version = "0.8", optional = true }
 tracing-subscriber = { version = "0.3", features = [
@@ -110,7 +110,7 @@ tokio = { version = "1", features = [
     "signal",
     "process",
 ] }
-tokio-websockets = { version = "0.11.3", features = ["rustls-bring-your-own-connector", "ring", "getrandom", "rand", "client"] }
+tokio-websockets = { version = "0.12", features = ["rustls-bring-your-own-connector", "ring", "getrandom", "rand", "client"] }
 
 # wasm-in-browser dependencies
 [target.'cfg(all(target_family = "wasm", target_os = "unknown"))'.dependencies]

iroh-relay/Cargo.toml

@@ -25,11 +25,7 @@ use url::Url;
 pub use self::conn::{ReceivedMessage, RecvError, SendError, SendMessage};
 #[cfg(not(wasm_browser))]
 use crate::dns::{DnsError, DnsResolver};
-use crate::{
-    http::{Protocol, RELAY_PATH},
-    protos::relay::SendError as SendRelayError,
-    KeyCache,
-};
+use crate::{http::RELAY_PATH, protos::relay::SendError as SendRelayError, KeyCache};
 
 pub(crate) mod conn;
 #[cfg(not(wasm_browser))]
@@ -125,8 +121,6 @@ pub struct ClientBuilder {
     is_prober: bool,
     /// Server url.
     url: RelayUrl,
-    /// Relay protocol
-    protocol: Protocol,
     /// Allow self-signed certificates from relay servers
     #[cfg(any(test, feature = "test-utils"))]
     insecure_skip_cert_verify: bool,
@@ -153,9 +147,6 @@ impl ClientBuilder {
             is_prober: false,
             url: url.into(),
 
-            // Resolves to websockets in browsers and relay otherwise
-            protocol: Protocol::default(),
-
             #[cfg(any(test, feature = "test-utils"))]
             insecure_skip_cert_verify: false,
 
@@ -167,13 +158,6 @@ impl ClientBuilder {
         }
     }
 
-    /// Sets whether to connect to the relay via websockets or not.
-    /// Set to use non-websocket, normal relaying by default.
-    pub fn protocol(mut self, protocol: Protocol) -> Self {
-        self.protocol = protocol;
-        self
-    }
-
     /// Returns if we should prefer ipv6
     /// it replaces the relayhttp.AddressFamilySelector we pass
     /// It provides the hint as to whether in an IPv4-vs-IPv6 race that
@@ -216,41 +200,97 @@ impl ClientBuilder {
     }
 
     /// Establishes a new connection to the relay server.
+    #[cfg(not(wasm_browser))]
     pub async fn connect(&self) -> Result<Client, ConnectError> {
-        let (conn, local_addr) = match self.protocol {
-            #[cfg(wasm_browser)]
-            Protocol::Websocket => {
-                let conn = self.connect_ws().await?;
-                let local_addr = None;
-                (conn, local_addr)
-            }
-            #[cfg(not(wasm_browser))]
-            Protocol::Websocket => {
-                let (conn, local_addr) = self.connect_ws().await?;
-                (conn, Some(local_addr))
-            }
-            #[cfg(not(wasm_browser))]
-            Protocol::Relay => {
-                let (conn, local_addr) = self.connect_relay().await?;
-                (conn, Some(local_addr))
+        use tls::MaybeTlsStreamBuilder;
+
+        use crate::protos::relay::MAX_FRAME_SIZE;
+
+        let mut dial_url = (*self.url).clone();
+        dial_url.set_path(RELAY_PATH);
+        // The relay URL is exchanged with the http(s) scheme in tickets and similar.
+        // We need to use the ws:// or wss:// schemes when connecting with websockets, though.
+        dial_url
+            .set_scheme(match self.url.scheme() {
+                "http" => "ws",
+                "ws" => "ws",
+                _ => "wss",
+            })
+            .map_err(|_| {
+                InvalidWebsocketUrlSnafu {
+                    url: dial_url.clone(),
+                }
+                .build()
+            })?;
+
+        debug!(%dial_url, "Dialing relay by websocket");
+
+        #[allow(unused_mut)]
+        let mut builder = MaybeTlsStreamBuilder::new(dial_url.clone(), self.dns_resolver.clone())
+            .prefer_ipv6(self.prefer_ipv6())
+            .proxy_url(self.proxy_url.clone());
+
+        #[cfg(any(test, feature = "test-utils"))]
+        if self.insecure_skip_cert_verify {
+            builder = builder.insecure_skip_cert_verify(self.insecure_skip_cert_verify);
+        }
+
+        let stream = builder.connect().await?;
+        let local_addr = stream
+            .as_ref()
+            .local_addr()
+            .map_err(|_| NoLocalAddrSnafu.build())?;
+        let (conn, response) = tokio_websockets::ClientBuilder::new()
+            .uri(dial_url.as_str())
+            .map_err(|_| {
+                InvalidRelayUrlSnafu {
+                    url: dial_url.clone(),
+                }
+                .build()
+            })?
+            .limits(tokio_websockets::Limits::default().max_payload_len(Some(MAX_FRAME_SIZE)))
+            .connect_on(stream)
+            .await?;
+
+        if response.status() != hyper::StatusCode::SWITCHING_PROTOCOLS {
+            UnexpectedUpgradeStatusSnafu {
+                code: response.status(),
             }
-            #[cfg(wasm_browser)]
-            Protocol::Relay => return Err(RelayProtoNotAvailableSnafu.build()),
-        };
+            .fail()?;
+        }
+
+        let conn = Conn::new(conn, self.key_cache.clone(), &self.secret_key).await?;
 
         event!(
             target: "events.net.relay.connected",
             Level::DEBUG,
             url = %self.url,
-            protocol = ?self.protocol,
         );
 
         trace!("connect done");
-        Ok(Client { conn, local_addr })
+
+        Ok(Client {
+            conn,
+            local_addr: Some(local_addr),
+        })
+    }
+
+    /// Reports whether IPv4 dials should be slightly
+    /// delayed to give IPv6 a better chance of winning dial races.
+    /// Implementations should only return true if IPv6 is expected
+    /// to succeed. (otherwise delaying IPv4 will delay the connection
+    /// overall)
+    #[cfg(not(wasm_browser))]
+    fn prefer_ipv6(&self) -> bool {
+        match self.address_family_selector {
+            Some(ref selector) => selector(),
+            None => false,
+        }
     }
 
+    /// Establishes a new connection to the relay server.
     #[cfg(wasm_browser)]
-    async fn connect_ws(&self) -> Result<Conn, ConnectError> {
+    pub async fn connect(&self) -> Result<Client, ConnectError> {
         let mut dial_url = (*self.url).clone();
         dial_url.set_path(RELAY_PATH);
         // The relay URL is exchanged with the http(s) scheme in tickets and similar.
@@ -271,9 +311,20 @@ impl ClientBuilder {
         debug!(%dial_url, "Dialing relay by websocket");
 
         let (_, ws_stream) = ws_stream_wasm::WsMeta::connect(dial_url.as_str(), None).await?;
-        let conn =
-            Conn::new_ws_browser(ws_stream, self.key_cache.clone(), &self.secret_key).await?;
-        Ok(conn)
+        let conn = Conn::new(ws_stream, self.key_cache.clone(), &self.secret_key).await?;
+
+        event!(
+            target: "events.net.relay.connected",
+            Level::DEBUG,
+            url = %self.url,
+        );
+
+        trace!("connect done");
+
+        Ok(Client {
+            conn,
+            local_addr: None,
+        })
     }
 }