feat: select a working iroh relay among all candidates from published relays

src/config.rs

@@ -19,7 +19,7 @@ use crate::log::LogExt;
 use crate::mimefactory::RECOMMENDED_FILE_SIZE;
 use crate::sync::{self, Sync::*, SyncData};
 use crate::tools::{get_abs_path, time};
-use crate::transport::{add_pseudo_transport, send_sync_transports};
+use crate::transport::{add_pseudo_transport, published_transports, send_sync_transports};
 use crate::{constants, stats};
 
 /// The available configuration keys.
@@ -946,16 +946,11 @@ impl Context {
     /// Returns all published self addresses, newest first.
     /// See `[Context::set_transport_unpublished]`
     pub(crate) async fn get_published_self_addrs(&self) -> Result<Vec<String>> {
-        self.sql
-            .query_map_vec(
-                "SELECT addr FROM transports WHERE is_published=1 ORDER BY add_timestamp DESC, id DESC",
-                (),
-                |row| {
-                    let addr: String = row.get(0)?;
-                    Ok(addr)
-                },
-            )
-            .await
+        Ok(published_transports(self)
+            .await?
+            .into_iter()
+            .map(|(addr, _)| addr)
+            .collect())
     }
 
     /// Returns all published secondary self addresses.

src/context.rs

@@ -1,15 +1,16 @@
 //! Context module.
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::ffi::OsString;
 use std::ops::Deref;
 use std::path::{Path, PathBuf};
-use std::sync::atomic::AtomicBool;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Arc, OnceLock, Weak};
 use std::time::Duration;
 
 use anyhow::{Result, bail, ensure};
 use async_channel::{self as channel, Receiver, Sender};
+use iroh_gossip::proto::TopicId;
 use pgp::composed::SignedPublicKey;
 use ratelimit::Ratelimit;
 use tokio::sync::{Mutex, Notify, RwLock};
@@ -313,7 +314,16 @@ pub struct InnerContext {
     pub(crate) spki_hash_store: SpkiHashStore,
 
     /// Iroh for realtime peer channels.
-    pub(crate) iroh: Arc<RwLock<Option<Iroh>>>,
+    pub(crate) iroh: RwLock<Option<Arc<Iroh>>>,
+
+    /// Mutex to serialize initialization and closing of [`Self::iroh`].
+    pub(crate) iroh_init_mutex: Mutex<()>,
+
+    /// Incremented on every [`Context::stop_io`] call to detect racing iroh initialization.
+    pub(crate) io_stop_count: AtomicUsize,
+
+    /// Topics left so that a racing join does not re-open their channel.
+    pub(crate) left_topics: parking_lot::Mutex<HashSet<TopicId>>,
 
     /// The own fingerprint, if it was computed already.
     /// tokio::sync::OnceCell would be possible to use, but overkill for our usecase;
@@ -502,7 +512,10 @@ impl Context {
             push_subscriber,
             tls_session_store: TlsSessionStore::new(),
             spki_hash_store: SpkiHashStore::new(),
-            iroh: Arc::new(RwLock::new(None)),
+            iroh: RwLock::new(None),
+            iroh_init_mutex: Mutex::new(()),
+            io_stop_count: AtomicUsize::new(0),
+            left_topics: parking_lot::Mutex::new(HashSet::new()),
             self_fingerprint: OnceLock::new(),
             self_public_key: Mutex::new(None),
             published_connectivities: parking_lot::Mutex::new(Vec::new()),
@@ -534,7 +547,9 @@ impl Context {
 
     /// Stops the IO scheduler.
     pub async fn stop_io(&self) {
+        self.io_stop_count.fetch_add(1, Ordering::Relaxed);
         self.scheduler.stop(self).await;
+        let _guard = self.iroh_init_mutex.lock().await;
         if let Some(iroh) = self.iroh.write().await.take() {
             // Close all QUIC connections.
 
@@ -558,7 +573,7 @@ impl Context {
 
     /// Indicate that the network likely has come back.
     pub async fn maybe_network(&self) {
-        if let Some(ref iroh) = *self.iroh.read().await {
+        if let Some(iroh) = self.iroh.read().await.clone() {
             iroh.network_change().await;
         }
         self.scheduler.maybe_network().await;

src/mimefactory.rs

@@ -1914,14 +1914,9 @@ impl MimeFactory {
             }
             SystemMessage::IrohNodeAddr => {
                 let node_addr = context
-                    .get_or_try_init_peer_channel()
+                    .get_active_or_init_iroh()
                     .await?
-                    .get_node_addr()
-                    .await?;
-
-                // We should not send `null` as relay URL
-                // as this is the only way to reach the node.
-                debug_assert!(node_addr.relay_url().is_some());
+                    .get_relay_node_addr()?;
                 headers.push((
                     HeaderDef::IrohNodeAddr.into(),
                     mail_builder::headers::text::Text::new(serde_json::to_string(&node_addr)?)

src/net/http.rs

@@ -52,6 +52,7 @@ async fn get_http_sender<B>(
     context: &Context,
     parsed_url: hyper::Uri,
     strict_tls: bool,
+    use_proxy: bool,
 ) -> Result<hyper::client::conn::http1::SendRequest<B>>
 where
     B: hyper::body::Body + 'static + Send,
@@ -60,7 +61,11 @@ where
 {
     let scheme = parsed_url.scheme_str().context("URL has no scheme")?;
     let host = parsed_url.host().context("URL has no host")?;
-    let proxy_config_opt = ProxyConfig::load(context).await?;
+    let proxy_config_opt = if use_proxy {
+        ProxyConfig::load(context).await?
+    } else {
+        None
+    };
 
     let stream: Box<dyn SessionStream> = match scheme {
         "http" => {
@@ -279,7 +284,7 @@ async fn fetch_url(context: &Context, original_url: &str, strict_tls: bool) -> R
             .parse::<hyper::Uri>()
             .with_context(|| format!("Failed to parse URL {url:?}"))?;
 
-        let mut sender = get_http_sender(context, parsed_url.clone(), strict_tls).await?;
+        let mut sender = get_http_sender(context, parsed_url.clone(), strict_tls, true).await?;
         let authority = parsed_url
             .authority()
             .context("URL has no authority")?
@@ -399,6 +404,34 @@ pub(crate) async fn read_url_blob_with_tls(
     Ok(response)
 }
 
+/// Probes an iroh relay URL with a non-cached GET request,
+/// failing unless a successful response status is received.
+pub(crate) async fn probe_iroh_url(context: &Context, url: &str) -> Result<()> {
+    let parsed_url = url
+        .parse::<hyper::Uri>()
+        .with_context(|| format!("Failed to parse URL {url:?}"))?;
+
+    // Connects directly, proxy off, like iroh does.
+    let strict_tls = true;
+    let use_proxy = false;
+    let mut sender = get_http_sender(context, parsed_url.clone(), strict_tls, use_proxy).await?;
+    let authority = parsed_url
+        .authority()
+        .context("URL has no authority")?
+        .clone();
+    let req = hyper::Request::get(origin_form(&parsed_url))
+        .header(hyper::header::HOST, authority.as_str())
+        .body(http_body_util::Empty::<Bytes>::new())?;
+
+    let response = sender.send_request(req).await?;
+    let status = response.status();
+    if !status.is_success() {
+        bail!("The server returned a non-successful response code: {status}");
+    }
+
+    Ok(())
+}
+
 /// Sends an empty POST request to the URL.
 ///
 /// Returns response text and whether request was successful or not.
@@ -413,7 +446,7 @@ pub(crate) async fn post_empty(context: &Context, url: &str) -> Result<(String,
         bail!("POST requests to non-HTTPS URLs are not allowed");
     }
 
-    let mut sender = get_http_sender(context, parsed_url.clone(), true).await?;
+    let mut sender = get_http_sender(context, parsed_url.clone(), true, true).await?;
     let authority = parsed_url
         .authority()
         .context("URL has no authority")?