feat: relay only configuration

This is a next step into the world of configurable transports. We now
allow disabling the IP based transports entirely.
Internally this starts to prepare for a world where the user can
configure multiple different transports, IP, relay and others in the
future.

Closes #2957

Author: dignifiedquire

iroh/bench/src/iroh.rs

@@ -6,7 +6,7 @@ use std::{
 use bytes::Bytes;
 use iroh::{
     Endpoint, EndpointAddr, RelayMode, RelayUrl,
-    endpoint::{Connection, ConnectionError, RecvStream, SendStream, TransportConfig},
+    endpoint::{Connection, ConnectionError, QuinnTransportConfig, RecvStream, SendStream},
 };
 use n0_error::{Result, StackResultExt, StdResultExt};
 use tracing::{trace, warn};
@@ -126,10 +126,10 @@ pub async fn connect_client(
     Ok((endpoint, connection))
 }
 
-pub fn transport_config(max_streams: usize, initial_mtu: u16) -> TransportConfig {
+pub fn transport_config(max_streams: usize, initial_mtu: u16) -> QuinnTransportConfig {
     // High stream windows are chosen because the amount of concurrent streams
     // is configurable as a parameter.
-    let mut config = TransportConfig::default();
+    let mut config = QuinnTransportConfig::default();
     config.max_concurrent_uni_streams(max_streams.try_into().unwrap());
     config.initial_mtu(initial_mtu);
 

iroh/src/endpoint.rs

@@ -56,7 +56,7 @@ pub use quinn::{
     ConnectionClose, ConnectionError, ConnectionStats, MtuDiscoveryConfig, OpenBi, OpenUni,
     PathStats, ReadDatagram, ReadError, ReadExactError, ReadToEndError, RecvStream, ResetError,
     RetryError, SendDatagramError, SendStream, ServerConfig, StoppedError, StreamId,
-    TransportConfig, VarInt, WeakConnectionHandle, WriteError,
+    TransportConfig as QuinnTransportConfig, VarInt, WeakConnectionHandle, WriteError,
 };
 pub use quinn_proto::{
     FrameStats, TransportError, TransportErrorCode, UdpStats, Written,
@@ -72,6 +72,7 @@ pub use self::connection::{
     Incoming, IncomingZeroRttConnection, OutgoingZeroRttConnection, RemoteEndpointIdError,
     ZeroRttStatus,
 };
+pub use crate::magicsock::transports::TransportConfig;
 
 /// The delay to fall back to discovery when direct addresses fail.
 ///
@@ -101,7 +102,6 @@ pub enum PathSelection {
 #[derive(Debug)]
 pub struct Builder {
     secret_key: Option<SecretKey>,
-    relay_mode: RelayMode,
     alpn_protocols: Vec<Vec<u8>>,
     transport_config: quinn::TransportConfig,
     keylog: bool,
@@ -112,13 +112,27 @@ pub struct Builder {
     dns_resolver: Option<DnsResolver>,
     #[cfg(any(test, feature = "test-utils"))]
     insecure_skip_relay_cert_verify: bool,
-    addr_v4: Option<SocketAddrV4>,
-    addr_v6: Option<SocketAddrV6>,
+    transports: Vec<TransportConfig>,
     #[cfg(any(test, feature = "test-utils"))]
     path_selection: PathSelection,
     max_tls_tickets: usize,
 }
 
+impl From<RelayMode> for Option<TransportConfig> {
+    fn from(mode: RelayMode) -> Self {
+        match mode {
+            RelayMode::Disabled => None,
+            RelayMode::Default => Some(TransportConfig::Relay {
+                relay_map: mode.relay_map(),
+            }),
+            RelayMode::Staging => Some(TransportConfig::Relay {
+                relay_map: mode.relay_map(),
+            }),
+            RelayMode::Custom(relay_map) => Some(TransportConfig::Relay { relay_map }),
+        }
+    }
+}
+
 impl Builder {
     // The ordering of public methods is reflected directly in the documentation.  This is
     // roughly ordered by what is most commonly needed by users.
@@ -140,9 +154,18 @@ impl Builder {
     pub fn empty(relay_mode: RelayMode) -> Self {
         let mut transport_config = quinn::TransportConfig::default();
         transport_config.keep_alive_interval(Some(Duration::from_secs(1)));
+
+        let mut transports = vec![
+            #[cfg(not(wasm_browser))]
+            TransportConfig::default_ipv4(),
+            #[cfg(not(wasm_browser))]
+            TransportConfig::default_ipv6(),
+        ];
+        if let Some(relay) = relay_mode.into() {
+            transports.push(relay);
+        }
         Self {
             secret_key: Default::default(),
-            relay_mode,
             alpn_protocols: Default::default(),
             transport_config: quinn::TransportConfig::default(),
             keylog: Default::default(),
@@ -153,11 +176,10 @@ impl Builder {
             dns_resolver: None,
             #[cfg(any(test, feature = "test-utils"))]
             insecure_skip_relay_cert_verify: false,
-            addr_v4: None,
-            addr_v6: None,
             #[cfg(any(test, feature = "test-utils"))]
             path_selection: PathSelection::default(),
             max_tls_tickets: DEFAULT_MAX_TLS_TICKETS,
+            transports,
         }
     }
 
@@ -166,7 +188,6 @@ impl Builder {
     /// Binds the magic endpoint.
     pub async fn bind(mut self) -> Result<Endpoint, BindError> {
         let mut rng = rand::rng();
-        let relay_map = self.relay_mode.relay_map();
         let secret_key = self
             .secret_key
             .unwrap_or_else(move || SecretKey::generate(&mut rng));
@@ -194,10 +215,8 @@ impl Builder {
         let metrics = EndpointMetrics::default();
 
         let msock_opts = magicsock::Options {
-            addr_v4: self.addr_v4,
-            addr_v6: self.addr_v6,
+            transports: self.transports,
             secret_key,
-            relay_map,
             discovery_user_data: self.discovery_user_data,
             proxy_url: self.proxy_url,
             #[cfg(not(wasm_browser))]
@@ -230,25 +249,46 @@ impl Builder {
 
     // # The very common methods everyone basically needs.
 
-    /// Sets the IPv4 bind address.
+    /// Adds an IP transport, binding to the provided IPv4 address.
+    ///
+    /// If you want to remove the default transports, make sure to call `clear_ip` first.
     ///
     /// Setting the port to `0` will use a random port.
     /// If the port specified is already in use, it will fallback to choosing a random port.
-    ///
-    /// By default will use `0.0.0.0:0` to bind to.
-    pub fn bind_addr_v4(mut self, addr: SocketAddrV4) -> Self {
-        self.addr_v4.replace(addr);
+    #[cfg(not(wasm_browser))]
+    pub fn bind_addr_v4(mut self, bind_addr: SocketAddrV4) -> Self {
+        self.transports.push(TransportConfig::Ip {
+            bind_addr: bind_addr.into(),
+        });
         self
     }
 
-    /// Sets the IPv6 bind address.
+    /// Adds an IP transport, binding to the provided IPv6 address.
+    ///
+    /// If you want to remove the default transports, make sure to call `clear_ip` first.
     ///
     /// Setting the port to `0` will use a random port.
     /// If the port specified is already in use, it will fallback to choosing a random port.
-    ///
-    /// By default will use `[::]:0` to bind to.
-    pub fn bind_addr_v6(mut self, addr: SocketAddrV6) -> Self {
-        self.addr_v6.replace(addr);
+    #[cfg(not(wasm_browser))]
+    pub fn bind_addr_v6(mut self, bind_addr: SocketAddrV6) -> Self {
+        self.transports.push(TransportConfig::Ip {
+            bind_addr: bind_addr.into(),
+        });
+        self
+    }
+
+    /// Removes all IP based transports
+    #[cfg(not(wasm_browser))]
+    pub fn clear_ip_transports(mut self) -> Self {
+        self.transports
+            .retain(|t| !matches!(t, TransportConfig::Ip { .. }));
+        self
+    }
+
+    /// Removes all relay based transports
+    pub fn clear_relay_transports(mut self) -> Self {
+        self.transports
+            .retain(|t| !matches!(t, TransportConfig::Relay { .. }));
         self
     }
 
@@ -294,7 +334,24 @@ impl Builder {
     /// [crate docs]: crate
     /// [number 0]: https://n0.computer
     pub fn relay_mode(mut self, relay_mode: RelayMode) -> Self {
-        self.relay_mode = relay_mode;
+        let transport: Option<_> = relay_mode.into();
+        match transport {
+            Some(transport) => {
+                if let Some(og) = self
+                    .transports
+                    .iter_mut()
+                    .find(|t| matches!(t, TransportConfig::Relay { .. }))
+                {
+                    *og = transport;
+                } else {
+                    self.transports.push(transport);
+                }
+            }
+            None => {
+                self.transports
+                    .retain(|t| !matches!(t, TransportConfig::Relay { .. }));
+            }
+        }
         self
     }
 
@@ -865,8 +922,6 @@ impl Endpoint {
         let endpoint_id = self.id();
 
         watch_addrs.or(watch_relay).map(move |(addrs, relays)| {
-            debug_assert!(!addrs.is_empty(), "direct addresses must never be empty");
-
             EndpointAddr::from_parts(
                 endpoint_id,
                 relays
@@ -1252,7 +1307,7 @@ impl Endpoint {
 /// Options for the [`Endpoint::connect_with_opts`] function.
 #[derive(Default, Debug, Clone)]
 pub struct ConnectOptions {
-    transport_config: Option<Arc<TransportConfig>>,
+    transport_config: Option<Arc<quinn::TransportConfig>>,
     additional_alpns: Vec<Vec<u8>>,
 }
 
@@ -1266,7 +1321,7 @@ impl ConnectOptions {
     }
 
     /// Sets the QUIC transport config options for this connection.
-    pub fn with_transport_config(mut self, transport_config: Arc<TransportConfig>) -> Self {
+    pub fn with_transport_config(mut self, transport_config: Arc<quinn::TransportConfig>) -> Self {
         self.transport_config = Some(transport_config);
         self
     }
@@ -1340,6 +1395,7 @@ fn proxy_url_from_env() -> Option<Url> {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum RelayMode {
     /// Disable relay servers completely.
+    /// This means that neither listening nor dialing relays will be available.
     Disabled,
     /// Use the default relay map, with production relay servers from n0.
     ///
@@ -1837,6 +1893,103 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    #[traced_test]
+    async fn endpoint_two_relay_only_no_ip() -> Result {
+        // Connect two endpoints on the same network, via a relay server, without
+        // discovery.
+        let (relay_map, _relay_url, _relay_server_guard) = run_relay_server().await?;
+        let (node_addr_tx, node_addr_rx) = oneshot::channel();
+
+        #[instrument(name = "client", skip_all)]
+        async fn connect(
+            relay_map: RelayMap,
+            node_addr_rx: oneshot::Receiver<EndpointAddr>,
+        ) -> Result<quinn::ConnectionError> {
+            let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
+            let secret = SecretKey::generate(&mut rng);
+            let ep = Endpoint::builder()
+                .secret_key(secret)
+                .alpns(vec![TEST_ALPN.to_vec()])
+                .insecure_skip_relay_cert_verify(true)
+                .relay_mode(RelayMode::Custom(relay_map))
+                .clear_ip_transports() // disable direct
+                .bind()
+                .await?;
+            info!(me = %ep.id().fmt_short(), "client starting");
+            let dst = node_addr_rx.await.anyerr()?;
+
+            info!(me = %ep.id().fmt_short(), "client connecting");
+            let conn = ep.connect(dst, TEST_ALPN).await?;
+            let mut send = conn.open_uni().await.anyerr()?;
+            send.write_all(b"hello").await.anyerr()?;
+            let mut paths = conn.paths().stream();
+            info!("Waiting for connection");
+            'outer: while let Some(infos) = paths.next().await {
+                info!(?infos, "new PathInfos");
+                for info in infos {
+                    if info.is_ip() {
+                        panic!("should not happen: {:?}", info);
+                    }
+                    if info.is_relay() {
+                        break 'outer;
+                    }
+                }
+            }
+            info!("Have relay connection");
+            send.write_all(b"close please").await.anyerr()?;
+            send.finish().anyerr()?;
+            Ok(conn.closed().await)
+        }
+
+        #[instrument(name = "server", skip_all)]
+        async fn accept(
+            relay_map: RelayMap,
+            node_addr_tx: oneshot::Sender<EndpointAddr>,
+        ) -> Result {
+            let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(1u64);
+            let secret = SecretKey::generate(&mut rng);
+            let ep = Endpoint::builder()
+                .secret_key(secret)
+                .alpns(vec![TEST_ALPN.to_vec()])
+                .insecure_skip_relay_cert_verify(true)
+                .relay_mode(RelayMode::Custom(relay_map))
+                .clear_ip_transports()
+                .bind()
+                .await?;
+            ep.online().await;
+            let node_addr = ep.addr();
+            node_addr_tx.send(node_addr).unwrap();
+
+            info!(me = %ep.id().fmt_short(), "server starting");
+            let conn = ep.accept().await.anyerr()?.await.anyerr()?;
+            // let node_id = conn.remote_node_id()?;
+            // assert_eq!(node_id, src);
+            let mut recv = conn.accept_uni().await.anyerr()?;
+            let mut msg = [0u8; 5];
+            recv.read_exact(&mut msg).await.anyerr()?;
+            assert_eq!(&msg, b"hello");
+            info!("received hello");
+            let msg = recv.read_to_end(100).await.anyerr()?;
+            assert_eq!(msg, b"close please");
+            info!("received 'close please'");
+            // Dropping the connection closes it just fine.
+            Ok(())
+        }
+
+        let server_task = tokio::spawn(accept(relay_map.clone(), node_addr_tx));
+        let client_task = tokio::spawn(connect(relay_map, node_addr_rx));
+
+        server_task.await.anyerr()??;
+        let conn_closed = dbg!(client_task.await.anyerr()??);
+        assert!(matches!(
+            conn_closed,
+            ConnectionError::ApplicationClosed(quinn::ApplicationClose { .. })
+        ));
+
+        Ok(())
+    }
+
     #[tokio::test]
     #[traced_test]
     async fn endpoint_two_direct_add_relay() -> Result {