some wip

- iroh-net -> iroh
- iroh-gossip uses hex by default, use base32 manually to keep backwards compat
- use the new `iroh::protocol::Router` to manage the gossip integration

Cargo.toml

@@ -63,8 +63,9 @@ humansize = "2"
 hyper = "1"
 hyper-util = "0.1.10"
 image = { version = "0.25.5", default-features=false, features = ["gif", "jpeg", "ico", "png", "pnm", "webp", "bmp"] }
-iroh-gossip = { version = "0.28.1", default-features = false, features = ["net"] }
-iroh-net = { version = "0.28.1", default-features = false }
+iroh-gossip = { version = "0.29", default-features = false, features = ["net"] }
+iroh = { version = "0.29", default-features = false }
+iroh-base = { version = "0.29", features = ["base32"] }
 kamadak-exif = "0.6.1"
 lettre_email = { git = "https://github.com/deltachat/lettre", branch = "master" }
 libc = { workspace = true }
@@ -122,6 +123,7 @@ proptest = { version = "1", default-features = false, features = ["std"] }
 tempfile = { workspace = true }
 testdir = "0.9.0"
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+tracing-subscriber.workspace = true
 
 [workspace]
 members = [

deny.toml

@@ -12,10 +12,6 @@ ignore = [
     # Unmaintained encoding
     "RUSTSEC-2021-0153",
 
-    # Unmaintained proc-macro-error
-    # <https://rustsec.org/advisories/RUSTSEC-2024-0370>
-    "RUSTSEC-2024-0370",
-
     # Unmaintained instant
     "RUSTSEC-2024-0384",
 ]
@@ -36,9 +32,13 @@ skip = [
      { name = "fiat-crypto", version = "0.1.20" },
      { name = "futures-lite", version = "1.13.0" },
      { name = "getrandom", version = "<0.2" },
+     { name = "hostname", version = "0.3.1" },
      { name = "http", version = "0.2.12" },
      { name = "idna", version = "0.5.0" },
+     { name = "netlink-packet-route", version = "0.17.1" },
+     { name = "netlink-packet-route", version = "0.21.0" },
      { name = "nix", version = "0.26.4" },
+     { name = "nix", version = "0.27.1" },
      { name = "num_enum_derive", version = "0.5.11" },
      { name = "num_enum", version = "0.5.11" },
      { name = "proc-macro-crate", version = "1.3.1" },
@@ -49,10 +49,16 @@ skip = [
      { name = "redox_syscall", version = "0.3.5" },
      { name = "regex-automata", version = "0.1.10" },
      { name = "regex-syntax", version = "0.6.29" },
+     { name = "rtnetlink", version = "0.13.1" },
+     { name = "strsim", version = "0.10.0" },
      { name = "sync_wrapper", version = "0.1.2" },
      { name = "syn", version = "1.0.109" },
+     { name = "thiserror-impl", version = "1.0.69" },
+     { name = "thiserror", version = "1.0.69" },
      { name = "time", version = "<0.3" },
+     { name = "tokio-tungstenite", version = "0.21.0" },
      { name = "toml_edit", version = "0.19.15" },
+     { name = "tungstenite", version = "0.21.0" },
      { name = "wasi", version = "<0.11" },
      { name = "windows_aarch64_gnullvm", version = "<0.52" },
      { name = "windows_aarch64_msvc", version = "<0.52" },

src/imex/transfer.rs

@@ -33,8 +33,7 @@ use std::task::Poll;
 
 use anyhow::{bail, format_err, Context as _, Result};
 use futures_lite::FutureExt;
-use iroh_net::relay::RelayMode;
-use iroh_net::Endpoint;
+use iroh::{Endpoint, RelayMode};
 use tokio::fs;
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
@@ -65,11 +64,11 @@ const BACKUP_ALPN: &[u8] = b"/deltachat/backup";
 /// task use the [`Context::stop_ongoing`] mechanism.
 #[derive(Debug)]
 pub struct BackupProvider {
-    /// iroh-net endpoint.
+    /// iroh endpoint.
     _endpoint: Endpoint,
 
-    /// iroh-net address.
-    node_addr: iroh_net::NodeAddr,
+    /// iroh address.
+    node_addr: iroh::NodeAddr,
 
     /// Authentication token that should be submitted
     /// to retrieve the backup.
@@ -162,7 +161,7 @@ impl BackupProvider {
 
     async fn handle_connection(
         context: Context,
-        conn: iroh_net::endpoint::Connecting,
+        conn: iroh::endpoint::Connecting,
         auth_token: String,
         dbfile: Arc<TempPathGuard>,
     ) -> Result<()> {
@@ -291,7 +290,7 @@ impl Future for BackupProvider {
 
 pub async fn get_backup2(
     context: &Context,
-    node_addr: iroh_net::NodeAddr,
+    node_addr: iroh::NodeAddr,
     auth_token: String,
 ) -> Result<()> {
     let relay_mode = RelayMode::Disabled;
@@ -337,7 +336,7 @@ pub async fn get_backup2(
 /// This is a long running operation which will return only when completed.
 ///
 /// Using [`Qr`] as argument is a bit odd as it only accepts specific variant of it.  It
-/// does avoid having [`iroh_net::NodeAddr`] in the primary API however, without
+/// does avoid having [`iroh::NodeAddr`] in the primary API however, without
 /// having to revert to untyped bytes.
 pub async fn get_backup(context: &Context, qr: Qr) -> Result<()> {
     match qr {

src/peer_channels.rs

@@ -26,15 +26,14 @@
 use anyhow::{anyhow, bail, Context as _, Result};
 use email::Header;
 use futures_lite::StreamExt;
+use iroh::key::{PublicKey, SecretKey};
+use iroh::{Endpoint, NodeAddr, NodeId, RelayMap, RelayMode, RelayUrl};
 use iroh_gossip::net::{Event, Gossip, GossipEvent, JoinOptions, GOSSIP_ALPN};
 use iroh_gossip::proto::TopicId;
-use iroh_net::key::{PublicKey, SecretKey};
-use iroh_net::relay::{RelayMap, RelayUrl};
-use iroh_net::{relay::RelayMode, Endpoint};
-use iroh_net::{NodeAddr, NodeId};
 use parking_lot::Mutex;
 use std::collections::{BTreeSet, HashMap};
 use std::env;
+use std::sync::Arc;
 use tokio::sync::{oneshot, RwLock};
 use tokio::task::JoinHandle;
 use url::Url;
@@ -54,11 +53,11 @@ const PUBLIC_KEY_STUB: &[u8] = "static_string".as_bytes();
 /// Store iroh peer channels for the context.
 #[derive(Debug)]
 pub struct Iroh {
-    /// [Endpoint] needed for iroh peer channels.
-    pub(crate) endpoint: Endpoint,
+    /// iroh router  needed for iroh peer channels.
+    pub(crate) router: iroh::protocol::Router,
 
     /// [Gossip] needed for iroh peer channels.
-    pub(crate) gossip: Gossip,
+    pub(crate) gossip: Arc<Gossip>,
 
     /// Sequence numbers for gossip channels.
     pub(crate) sequence_numbers: Mutex<HashMap<TopicId, i32>>,
@@ -75,15 +74,12 @@ pub struct Iroh {
 impl Iroh {
     /// Notify the endpoint that the network has changed.
     pub(crate) async fn network_change(&self) {
-        self.endpoint.network_change().await
+        self.router.endpoint().network_change().await
     }
 
     /// Closes the QUIC endpoint.
     pub(crate) async fn close(self) -> Result<()> {
-        self.endpoint
-            .close(0u32.into(), b"")
-            .await
-            .context("Closing iroh endpoint failed")
+        self.router.shutdown().await.context("Closing iroh failed")
     }
 
     /// Join a topic and create the subscriber loop for it.
@@ -121,7 +117,7 @@ impl Iroh {
         // Inform iroh of potentially new node addresses
         for node_addr in &peers {
             if !node_addr.info.is_empty() {
-                self.endpoint.add_node_addr(node_addr.clone())?;
+                self.router.endpoint().add_node_addr(node_addr.clone())?;
             }
         }
 
@@ -148,7 +144,7 @@ impl Iroh {
     pub async fn maybe_add_gossip_peers(&self, topic: TopicId, peers: Vec<NodeAddr>) -> Result<()> {
         if self.iroh_channels.read().await.get(&topic).is_some() {
             for peer in &peers {
-                self.endpoint.add_node_addr(peer.clone())?;
+                self.router.endpoint().add_node_addr(peer.clone())?;
             }
 
             self.gossip.join_with_opts(
@@ -198,7 +194,7 @@ impl Iroh {
 
     /// Get the iroh [NodeAddr] without direct IP addresses.
     pub(crate) async fn get_node_addr(&self) -> Result<NodeAddr> {
-        let mut addr = self.endpoint.node_addr().await?;
+        let mut addr = self.router.endpoint().node_addr().await?;
         addr.info.direct_addresses = BTreeSet::new();
         Ok(addr)
     }
@@ -275,16 +271,19 @@ impl Context {
             max_message_size: 128 * 1024,
             ..Default::default()
         };
-        let gossip = Gossip::from_endpoint(endpoint.clone(), gossip_config, &my_addr.info);
+        let gossip = Arc::new(Gossip::from_endpoint(
+            endpoint.clone(),
+            gossip_config,
+            &my_addr.info,
+        ));
 
-        // spawn endpoint loop that forwards incoming connections to the gossiper
-        let context = self.clone();
-
-        // Shuts down on deltachat shutdown
-        tokio::spawn(endpoint_loop(context, endpoint.clone(), gossip.clone()));
+        let router = iroh::protocol::Router::builder(endpoint)
+            .accept(GOSSIP_ALPN, gossip.clone())
+            .spawn()
+            .await?;
 
         Ok(Iroh {
-            endpoint,
+            router,
             gossip,
             sequence_numbers: Mutex::new(HashMap::new()),
             iroh_channels: RwLock::new(HashMap::new()),
@@ -507,54 +506,13 @@ fn create_random_topic() -> TopicId {
 pub(crate) async fn create_iroh_header(ctx: &Context, msg_id: MsgId) -> Result<Header> {
     let topic = create_random_topic();
     insert_topic_stub(ctx, msg_id, topic).await?;
+    let topic_string = iroh_base::base32::fmt(topic.as_bytes());
     Ok(Header::new(
         HeaderDef::IrohGossipTopic.get_headername().to_string(),
-        topic.to_string(),
+        topic_string,
     ))
 }
 
-async fn endpoint_loop(context: Context, endpoint: Endpoint, gossip: Gossip) {
-    while let Some(conn) = endpoint.accept().await {
-        let conn = match conn.accept() {
-            Ok(conn) => conn,
-            Err(err) => {
-                warn!(context, "Failed to accept iroh connection: {err:#}.");
-                continue;
-            }
-        };
-        info!(context, "IROH_REALTIME: accepting iroh connection");
-        let gossip = gossip.clone();
-        let context = context.clone();
-        tokio::spawn(async move {
-            if let Err(err) = handle_connection(&context, conn, gossip).await {
-                warn!(context, "IROH_REALTIME: iroh connection error: {err}");
-            }
-        });
-    }
-}
-
-async fn handle_connection(
-    context: &Context,
-    mut conn: iroh_net::endpoint::Connecting,
-    gossip: Gossip,
-) -> anyhow::Result<()> {
-    let alpn = conn.alpn().await?;
-    let conn = conn.await?;
-    let peer_id = iroh_net::endpoint::get_remote_node_id(&conn)?;
-
-    match alpn.as_slice() {
-        GOSSIP_ALPN => gossip
-            .handle_connection(conn)
-            .await
-            .context(format!("Gossip connection to {peer_id} failed"))?,
-        _ => warn!(
-            context,
-            "Ignoring connection from {peer_id}: unsupported ALPN protocol"
-        ),
-    }
-    Ok(())
-}
-
 async fn subscribe_loop(
     context: &Context,
     mut stream: iroh_gossip::net::GossipReceiver,
@@ -605,6 +563,8 @@ async fn subscribe_loop(
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
     use super::*;
     use crate::{
         chat::send_msg,
@@ -783,6 +743,13 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_can_reconnect() {
+        tokio::time::timeout(Duration::from_secs(20), test_can_reconnect_impl())
+            .await
+            .unwrap();
+    }
+
+    async fn test_can_reconnect_impl() {
+        tracing_subscriber::fmt::init();
         let mut tcm = TestContextManager::new();
         let alice = &mut tcm.alice().await;
         let bob = &mut tcm.bob().await;
@@ -971,6 +938,7 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_parallel_connect() {
+        eprintln!("START-----");
         let mut tcm = TestContextManager::new();
         let alice = &mut tcm.alice().await;
         let bob = &mut tcm.bob().await;

src/qr.rs

@@ -112,7 +112,7 @@ pub enum Qr {
     /// Provides a backup that can be retrieved using iroh-net based backup transfer protocol.
     Backup2 {
         /// Iroh node address.
-        node_addr: iroh_net::NodeAddr,
+        node_addr: iroh::NodeAddr,
 
         /// Authentication token.
         auth_token: String,
@@ -644,7 +644,7 @@ fn decode_backup2(qr: &str) -> Result<Qr> {
         .split_once('&')
         .context("Backup QR code has no separator")?;
     let auth_token = auth_token.to_string();
-    let node_addr = serde_json::from_str::<iroh_net::NodeAddr>(node_addr)
+    let node_addr = serde_json::from_str::<iroh::NodeAddr>(node_addr)
         .context("Invalid node addr in backup QR code")?;
 
     Ok(Qr::Backup2 {

src/receive_imf.rs

@@ -1,7 +1,6 @@
 //! Internet Message Format reception pipeline.
 
 use std::collections::HashSet;
-use std::str::FromStr;
 
 use anyhow::{Context as _, Result};
 use deltachat_contact_tools::{addr_cmp, may_be_valid_addr, sanitize_single_line, ContactAddress};
@@ -1558,8 +1557,8 @@ INSERT INTO msgs
   (
     id,
     rfc724_mid, chat_id,
-    from_id, to_id, timestamp, timestamp_sent, 
-    timestamp_rcvd, type, state, msgrmsg, 
+    from_id, to_id, timestamp, timestamp_sent,
+    timestamp_rcvd, type, state, msgrmsg,
     txt, txt_normalized, subject, txt_raw, param, hidden,
     bytes, mime_headers, mime_compressed, mime_in_reply_to,
     mime_references, mime_modified, error, ephemeral_timer,
@@ -1652,7 +1651,10 @@ RETURNING id
         // check if any part contains a webxdc topic id
         if part.typ == Viewtype::Webxdc {
             if let Some(topic) = mime_parser.get_header(HeaderDef::IrohGossipTopic) {
-                let topic = TopicId::from_str(topic).context("wrong gossip topic header")?;
+                // default encoding of topic ids is `hex`.
+                let topic_raw: [u8; 32] =
+                    iroh_base::base32::parse_array(topic).context("wrong gossip topic header")?;
+                let topic = TopicId::from_bytes(topic_raw);
                 insert_topic_stub(context, *msg_id, topic).await?;
             } else {
                 warn!(context, "webxdc doesn't have a gossip topic")