start figuring out more details

Author: dignifiedquire

iroh/src/endpoint.rs

@@ -732,14 +732,13 @@ impl Endpoint {
             self.add_node_addr(node_addr.clone())?;
         }
         let node_id = node_addr.node_id;
-        let direct_addresses = node_addr.direct_addresses.clone();
         let relay_url = node_addr.relay_url.clone();
 
         // Get the mapped IPv6 address from the magic socket. Quinn will connect to this
         // address.  Start discovery for this node if it's enabled and we have no valid or
         // verified address information for this node.  Dropping the discovery cancels any
         // still running task.
-        let (mapped_addr, _discovery_drop_guard) = self
+        let (mapped_addr, direct_addresses, _discovery_drop_guard) = self
             .get_mapping_addr_and_maybe_start_discovery(node_addr)
             .await
             .context(NoAddressSnafu)?;
@@ -770,18 +769,26 @@ impl Endpoint {
         };
 
         // TODO: race available addresses, this is currently only using the relay addr to connect
-        let dest_addr = mapped_addr.private_socket_addr();
+        let dest_addr = if relay_url.is_none() && !direct_addresses.is_empty() {
+            direct_addresses[0]
+        } else {
+            mapped_addr.private_socket_addr()
+        };
         let server_name = &tls::name::encode(node_id);
         let connect = self
             .msock
             .endpoint()
             .connect_with(client_config, dest_addr, server_name)
             .context(QuinnSnafu)?;
 
+        let mut paths = direct_addresses;
+        paths.push(mapped_addr.private_socket_addr());
+
         Ok(Connecting {
             inner: connect,
             ep: self.clone(),
             remote_node_id: Some(node_id),
+            paths,
             _discovery_drop_guard,
         })
     }
@@ -1376,18 +1383,20 @@ impl Endpoint {
     async fn get_mapping_addr_and_maybe_start_discovery(
         &self,
         node_addr: NodeAddr,
-    ) -> Result<(NodeIdMappedAddr, Option<DiscoveryTask>), GetMappingAddressError> {
+    ) -> Result<(NodeIdMappedAddr, Vec<SocketAddr>, Option<DiscoveryTask>), GetMappingAddressError>
+    {
         let node_id = node_addr.node_id;
 
         // Only return a mapped addr if we have some way of dialing this node, in other
         // words, we have either a relay URL or at least one direct address.
         let addr = if self.msock.has_send_address(node_id) {
-            self.msock.get_mapping_addr(node_id)
+            let maddr = self.msock.get_mapping_addr(node_id);
+            maddr.map(|maddr| (maddr, self.msock.get_direct_addrs(node_id)))
         } else {
             None
         };
         match addr {
-            Some(addr) => {
+            Some((maddr, direct)) => {
                 // We have some way of dialing this node, but that doesn't actually mean
                 // we can actually connect to any of these addresses.
                 // Therefore, we will invoke the discovery service if we haven't received from the
@@ -1399,7 +1408,7 @@ impl Endpoint {
                 let discovery = DiscoveryTask::maybe_start_after_delay(self, node_id, delay)
                     .ok()
                     .flatten();
-                Ok((addr, discovery))
+                Ok((maddr, direct, discovery))
             }
 
             None => {
@@ -1414,7 +1423,8 @@ impl Endpoint {
                     .await
                     .context(get_mapping_address_error::DiscoverSnafu)?;
                 if let Some(addr) = self.msock.get_mapping_addr(node_id) {
-                    Ok((addr, Some(discovery)))
+                    let direct = self.msock.get_direct_addrs(node_id);
+                    Ok((addr, direct, Some(discovery)))
                 } else {
                     Err(get_mapping_address_error::NoAddressSnafu.build())
                 }
@@ -1643,6 +1653,8 @@ pub struct Connecting {
     inner: quinn::Connecting,
     ep: Endpoint,
     remote_node_id: Option<NodeId>,
+    /// Additional paths to open once a connection is created
+    paths: Vec<SocketAddr>,
     /// We run discovery as long as we haven't established a connection yet.
     #[debug("Option<DiscoveryTask>")]
     _discovery_drop_guard: Option<DiscoveryTask>,
@@ -1771,15 +1783,21 @@ impl Future for Connecting {
                 if let Some(remote) = *this.remote_node_id {
                     let weak_handle = conn.inner.weak_handle();
                     let path_events = conn.inner.path_events();
-                    this.ep
-                        .msock
-                        .register_connection(remote, weak_handle, path_events);
+                    this.ep.msock.register_connection(
+                        remote,
+                        weak_handle,
+                        path_events,
+                        this.paths.clone(),
+                    );
                 } else if let Ok(remote) = conn.remote_node_id() {
                     let weak_handle = conn.inner.weak_handle();
                     let path_events = conn.inner.path_events();
-                    this.ep
-                        .msock
-                        .register_connection(remote, weak_handle, path_events);
+                    this.ep.msock.register_connection(
+                        remote,
+                        weak_handle,
+                        path_events,
+                        this.paths.clone(),
+                    );
                 } else {
                     warn!("unable to determine node id for the remote");
                 }

iroh/src/magicsock.rs

@@ -295,8 +295,24 @@ impl MagicSock {
         remote: NodeId,
         conn: WeakConnectionHandle,
         mut path_events: tokio::sync::broadcast::Receiver<PathEvent>,
+        paths: Vec<SocketAddr>,
     ) {
         self.connection_map.insert(remote, conn);
+        task::spawn(async move {
+            let conn = conn.clone();
+            for addr in paths {
+                match conn.open_path(addr, quinn_proto::PathStatus::Backup).await {
+                    Ok(path) => {
+                        path.set_max_idle_timeout(Some(ENDPOINTS_FRESH_ENOUGH_DURATION))
+                            .ok();
+                        path.set_keep_alive_interval(Some(HEARTBEAT_INTERVAL)).ok();
+                    }
+                    Err(err) => {
+                        warn!("failed to open path {:?}", err);
+                    }
+                }
+            }
+        });
 
         // TODO: track task
         // TODO: find a good home for this
@@ -422,6 +438,10 @@ impl MagicSock {
         self.node_map.get_quic_mapped_addr_for_node_key(node_id)
     }
 
+    pub(crate) fn get_direct_addrs(&self, node_id: NodeId) -> Vec<SocketAddr> {
+        self.node_map.get_direct_addrs(node_id)
+    }
+
     /// Add addresses for a node to the magic socket's addresbook.
     #[instrument(skip_all)]
     pub fn add_node_addr(
@@ -1055,6 +1075,40 @@ impl MagicSock {
     }
 }
 
+/// Definies the translation of addresses in quinn land vs iroh land.
+///
+/// This is necessary, because quinn can only reason about `SocketAddr`s.
+#[derive(Clone, Debug)]
+pub(crate) enum MultipathMappedAddr {
+    /// Used for the initial connection.
+    /// - Only used for sending
+    /// - This means send on all known paths/transports
+    Mixed(NodeIdMappedAddr),
+    /// Relay based transport address
+    Relay(IpMappedAddr), // TODO: RelayMappedAddr?
+    /// IP based transport address
+    #[cfg(not(wasm_browser))]
+    Ip(SocketAddr),
+}
+
+impl From<SocketAddr> for MultipathMappedAddr {
+    fn from(value: SocketAddr) -> Self {
+        match value.ip() {
+            IpAddr::V4(_) => Self::Ip(value),
+            IpAddr::V6(addr) => {
+                if let Ok(node_id_mapped_addr) = NodeIdMappedAddr::try_from(addr) {
+                    return Self::Mixed(node_id_mapped_addr);
+                }
+                #[cfg(not(wasm_browser))]
+                if let Ok(ip_mapped_addr) = IpMappedAddr::try_from(addr) {
+                    return Self::Relay(ip_mapped_addr);
+                }
+                MappedAddr::Self(value)
+            }
+        }
+    }
+}
+
 #[derive(Clone, Debug)]
 enum MappedAddr {
     NodeId(NodeIdMappedAddr),
@@ -3187,17 +3241,18 @@ mod tests {
         let _accept_task = AbortOnDropHandle::new(accept_task);
 
         // Add an empty entry in the NodeMap of ep_1
-        msock_1.node_map.add_node_addr(
-            NodeAddr {
-                node_id: node_id_2,
-                relay_url: None,
-                direct_addresses: Default::default(),
-            },
-            Source::NamedApp {
-                name: "test".into(),
-            },
-            &msock_1.metrics.magicsock,
-        );
+        msock_1
+            .add_node_addr(
+                NodeAddr {
+                    node_id: node_id_2,
+                    relay_url: None,
+                    direct_addresses: Default::default(),
+                },
+                Source::NamedApp {
+                    name: "test".into(),
+                },
+            )
+            .unwrap();
         let addr_2 = msock_1.get_mapping_addr(node_id_2).unwrap();
 
         // Set a low max_idle_timeout so quinn gives up on this quickly and our test does
@@ -3224,24 +3279,25 @@ mod tests {
         info!("first connect timed out as expected");
 
         // Provide correct addressing information
-        msock_1.node_map.add_node_addr(
-            NodeAddr {
-                node_id: node_id_2,
-                relay_url: None,
-                direct_addresses: msock_2
-                    .direct_addresses()
-                    .initialized()
-                    .await
-                    .expect("no direct addrs")
-                    .into_iter()
-                    .map(|x| x.addr)
-                    .collect(),
-            },
-            Source::NamedApp {
-                name: "test".into(),
-            },
-            &msock_1.metrics.magicsock,
-        );
+        msock_1
+            .add_node_addr(
+                NodeAddr {
+                    node_id: node_id_2,
+                    relay_url: None,
+                    direct_addresses: msock_2
+                        .direct_addresses()
+                        .initialized()
+                        .await
+                        .expect("no direct addrs")
+                        .into_iter()
+                        .map(|x| x.addr)
+                        .collect(),
+                },
+                Source::NamedApp {
+                    name: "test".into(),
+                },
+            )
+            .unwrap();
 
         // We can now connect
         tokio::time::timeout(Duration::from_secs(10), async move {

iroh/src/magicsock/node_map.rs

@@ -180,6 +180,15 @@ impl NodeMap {
             .map(|ep| *ep.quic_mapped_addr())
     }
 
+    pub(super) fn get_direct_addrs(&self, node_key: NodeId) -> Vec<SocketAddr> {
+        self.inner
+            .lock()
+            .expect("poisoned")
+            .get(NodeStateKey::NodeId(node_key))
+            .map(|ep| ep.direct_addresses().map(Into::into).collect())
+            .unwrap_or_default()
+    }
+
     pub(super) fn handle_call_me_maybe(
         &self,
         sender: PublicKey,