fix: don't join a realtime topic that was just left

A concurrent join while the channel is left would register it afterwards
and keep iroh running although the webxdc app was closed.
The race window got wider with initialization probing relay candidates.

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")?

src/peer_channels.rs

@@ -23,31 +23,41 @@
 //!    (scoped per WebXDC app instance/message-id). The other peers can then join the gossip with `joinRealtimeChannel().setListener()`
 //!    and `joinRealtimeChannel().send()` just like the other peers.
 
-use anyhow::{Context as _, Result, anyhow, bail};
+use anyhow::{Context as _, Result, anyhow, bail, ensure};
 use data_encoding::BASE32_NOPAD;
 use futures_lite::StreamExt;
 use iroh::{Endpoint, NodeAddr, NodeId, PublicKey, RelayMode, RelayUrl, SecretKey};
 use iroh_gossip::net::{Event, GOSSIP_ALPN, Gossip, GossipEvent, JoinOptions};
 use iroh_gossip::proto::TopicId;
 use parking_lot::Mutex;
-use std::collections::{BTreeSet, HashMap};
+use std::collections::HashMap;
 use std::env;
+use std::sync::Arc;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
 use tokio::sync::{RwLock, oneshot};
 use tokio::task::JoinHandle;
+use tokio::time::timeout;
 use url::Url;
 
 use crate::EventType;
 use crate::chat::send_msg;
 use crate::config::Config;
 use crate::context::Context;
-use crate::log::warn;
+use crate::log::{LogExt, warn};
 use crate::message::{Message, MsgId, Viewtype};
 use crate::mimeparser::SystemMessage;
+use crate::net::http::probe_iroh_url;
+use crate::net::run_connection_attempts;
+use crate::transport::published_transports;
 
 /// The length of an ed25519 `PublicKey`, in bytes.
 const PUBLIC_KEY_LENGTH: usize = 32;
 const PUBLIC_KEY_STUB: &[u8] = "static_string".as_bytes();
 
+/// Timeout for probing an iroh relay candidate.
+const RELAY_PROBE_TIMEOUT: Duration = Duration::from_secs(10);
+
 /// Store Iroh peer channels for the context.
 #[derive(Debug)]
 pub struct Iroh {
@@ -67,6 +77,9 @@ pub struct Iroh {
     ///
     /// This is attached to every message to work around `iroh_gossip` deduplication.
     pub(crate) public_key: PublicKey,
+
+    /// Home relay URL probed successfully when Iroh was created.
+    working_relay_url: Option<RelayUrl>,
 }
 
 impl Iroh {
@@ -76,7 +89,7 @@ impl Iroh {
     }
 
     /// Closes the QUIC endpoint.
-    pub(crate) async fn close(self) -> Result<()> {
+    pub(crate) async fn close(&self) -> Result<()> {
         self.router.shutdown().await.context("Closing iroh failed")
     }
 
@@ -99,6 +112,10 @@ impl Iroh {
         // after we check that it does not exist and before we create a new one.
         // Otherwise we would receive every message twice or more times.
         let mut iroh_channels = self.iroh_channels.write().await;
+        ensure!(
+            !ctx.left_topics.lock().contains(&topic),
+            "Realtime channel for {msg_id} was left"
+        );
 
         if iroh_channels.contains_key(&topic) {
             return Ok(None);
@@ -184,16 +201,14 @@ impl Iroh {
         *entry
     }
 
-    /// Get the iroh [NodeAddr] without direct IP addresses.
-    ///
-    /// The address is guaranteed to have home relay URL set
-    /// as it is the only way to reach the node
-    /// without global discovery mechanisms.
-    pub(crate) async fn get_node_addr(&self) -> Result<NodeAddr> {
-        let mut addr = self.router.endpoint().node_addr().await?;
-        addr.direct_addresses = BTreeSet::new();
-        debug_assert!(addr.relay_url().is_some());
-        Ok(addr)
+    /// Returns the iroh [NodeAddr] with the working relay URL
+    /// and without direct IP addresses.
+    pub(crate) fn get_relay_node_addr(&self) -> Result<NodeAddr> {
+        let relay_url = self
+            .working_relay_url
+            .clone()
+            .context("No working iroh relay, peers cannot reach us")?;
+        Ok(NodeAddr::new(self.public_key).with_relay_url(relay_url))
     }
 
     /// Leave the realtime channel for a given topic.
@@ -231,28 +246,88 @@ impl ChannelState {
     }
 }
 
+/// Returns the URL to probe for the given iroh relay.
+///
+/// The probe path is appended to any path of the relay URL
+/// so that relays served under a path prefix can be probed too.
+fn relay_probe_url(relay_url: &Url) -> Result<Url> {
+    let mut probe_url = relay_url.clone();
+    probe_url
+        .path_segments_mut()
+        .map_err(|()| anyhow!("Relay URL {relay_url} cannot be a base"))?
+        .pop_if_empty()
+        .push("generate_204");
+    Ok(probe_url)
+}
+
+/// Selects a working iroh relay among the candidates.
+async fn select_iroh_relay(context: &Context, candidates: &[Url]) -> Result<Url> {
+    let probes = candidates.iter().cloned().map(|candidate| {
+        let context = context.clone();
+        async move {
+            let probe_target = relay_probe_url(&candidate)?;
+            timeout(
+                RELAY_PROBE_TIMEOUT,
+                probe_iroh_url(&context, probe_target.as_str()),
+            )
+            .await
+            .with_context(|| format!("Timeout probing iroh relay {candidate}"))?
+            .with_context(|| format!("Failed to probe iroh relay {candidate}"))?;
+            Ok(candidate)
+        }
+    });
+    let selected = run_connection_attempts(probes).await?;
+    info!(context, "Selected iroh relay {selected}.");
+    Ok(selected)
+}
+
+/// Returns the deduplicated iroh relay URLs announced by the published transports.
+async fn published_iroh_relays(context: &Context) -> Result<Vec<Url>> {
+    let published = published_transports(context).await?;
+    let metadata = context.metadata.read().await;
+    let mut relays: Vec<Url> = Vec::new();
+    for (_, transport_id) in published {
+        if let Some(url) = metadata
+            .get(&transport_id)
+            .and_then(|conf| conf.iroh_relay.clone())
+            && !relays.contains(&url)
+        {
+            relays.push(url);
+        }
+    }
+    Ok(relays)
+}
+
 impl Context {
-    /// Create iroh endpoint and gossip.
-    async fn init_peer_channels(&self) -> Result<Iroh> {
-        info!(self, "Initializing peer channels.");
+    /// Creates the iroh endpoint, gossip and router.
+    ///
+    /// Iroh is created even in the rare case
+    /// that no relay candidate works;
+    /// peers cannot reach us then,
+    /// but we can still dial out through their relays.
+    async fn init_iroh(&self) -> Result<Iroh> {
+        if !self.get_config_bool(Config::WebxdcRealtimeEnabled).await? {
+            bail!("Attempt to initialize Iroh when realtime is disabled");
+        }
+        info!(self, "Initializing Iroh for realtime channels.");
+        let relay_candidates = published_iroh_relays(self).await?;
+        let working_relay_url = if relay_candidates.is_empty() {
+            warn!(self, "No iroh relay, peers cannot reach us.");
+            None
+        } else {
+            select_iroh_relay(self, &relay_candidates)
+                .await
+                .context("No working iroh relay")
+                .log_err(self)
+                .ok()
+                .map(RelayUrl::from)
+        };
+        let relay_mode = match &working_relay_url {
+            Some(relay_url) => RelayMode::Custom(relay_url.clone().into()),
+            None => RelayMode::Disabled,
+        };
         let secret_key = SecretKey::generate(rand_old::rngs::OsRng);
         let public_key = secret_key.public();
-
-        let relay_mode = if let Some(relay_url) = self
-            .metadata
-            .read()
-            .await
-            .values()
-            .next()
-            .and_then(|conf| conf.iroh_relay.clone())
-        {
-            RelayMode::Custom(RelayUrl::from(relay_url).into())
-        } else {
-            // FIXME: this should be RelayMode::Disabled instead.
-            // Currently using default relays because otherwise Rust tests fail.
-            RelayMode::Default
-        };
-
         let endpoint = Box::pin(
             Endpoint::builder()
                 .tls_x509() // For compatibility with iroh <0.34.0
@@ -283,50 +358,75 @@ impl Context {
             sequence_numbers: Mutex::new(HashMap::new()),
             iroh_channels: RwLock::new(HashMap::new()),
             public_key,
+            working_relay_url,
         })
     }
 
-    /// Returns [`None`] if the peer channels has not been initialized.
-    pub async fn get_peer_channels(&self) -> Option<tokio::sync::RwLockReadGuard<'_, Iroh>> {
-        tokio::sync::RwLockReadGuard::<'_, std::option::Option<Iroh>>::try_map(
-            self.iroh.read().await,
-            |opt_iroh| opt_iroh.as_ref(),
-        )
-        .ok()
+    /// Returns iroh while it has a working relay
+    /// or channels through which peers may have been dialed.
+    async fn get_active_iroh(&self) -> Option<Arc<Iroh>> {
+        let iroh = self.iroh.read().await.clone()?;
+        if iroh.working_relay_url.is_some() || !iroh.iroh_channels.read().await.is_empty() {
+            Some(iroh)
+        } else {
+            None
+        }
     }
 
-    /// Get or initialize the iroh peer channel.
-    pub async fn get_or_try_init_peer_channel(
-        &self,
-    ) -> Result<tokio::sync::RwLockReadGuard<'_, Iroh>> {
-        if !self.get_config_bool(Config::WebxdcRealtimeEnabled).await? {
-            bail!("Attempt to initialize Iroh when realtime is disabled");
+    /// Returns active iroh, initializing it if necessary.
+    ///
+    /// Concurrent calls are serialized, and a call racing [`Context::stop_io`] fails
+    /// rather than leaving iroh running after the stop.
+    /// Inactive iroh is replaced, probing the relay candidates again.
+    pub(crate) async fn get_active_or_init_iroh(&self) -> Result<Arc<Iroh>> {
+        if let Some(iroh) = self.get_active_iroh().await {
+            return Ok(iroh);
         }
 
-        if let Some(lock) = self.get_peer_channels().await {
-            return Ok(lock);
+        let io_stop_count = self.io_stop_count.load(Ordering::Relaxed);
+        let _guard = self.iroh_init_mutex.lock().await;
+
+        if let Some(iroh) = self.get_active_iroh().await {
+            return Ok(iroh);
+        }
+        if self.io_stop_count.load(Ordering::Relaxed) != io_stop_count {
+            bail!("Io was stopped");
         }
 
-        let lock = self.iroh.write().await;
-        match tokio::sync::RwLockWriteGuard::<'_, std::option::Option<Iroh>>::try_downgrade_map(
-            lock,
-            |opt_iroh| opt_iroh.as_ref(),
-        ) {
-            Ok(lock) => Ok(lock),
-            Err(mut lock) => {
-                let iroh = self.init_peer_channels().await?;
-                *lock = Some(iroh);
-                tokio::sync::RwLockWriteGuard::<'_, std::option::Option<Iroh>>::try_downgrade_map(
-                    lock,
-                    |opt_iroh| opt_iroh.as_ref(),
-                )
-                .map_err(|_| anyhow!("Downgrade should succeed as we just stored `Some` value"))
-            }
+        // Close the inactive instance to probe the relay candidates anew,
+        // unless a concurrent task still uses it; share it with them then.
+        if let Some(iroh) = self.close_unused_iroh().await {
+            return Ok(iroh);
         }
+
+        let iroh = Arc::new(self.init_iroh().await?);
+        *self.iroh.write().await = Some(iroh.clone());
+        Ok(iroh)
+    }
+
+    /// Closes iroh if no channel and no concurrent task uses it,
+    /// so that it stops using the network until realtime is used again.
+    /// Returns the instance kept because it is still in use, if any.
+    ///
+    /// The next use initializes iroh again and probes the relay candidates,
+    /// so a relay that stopped working meanwhile is not used again.
+    pub(crate) async fn close_unused_iroh(&self) -> Option<Arc<Iroh>> {
+        let mut slot = self.iroh.write().await;
+        let iroh = slot.take()?;
+        // A reference besides ours means somebody is about to use iroh.
+        if Arc::strong_count(&iroh) > 1 || !iroh.iroh_channels.read().await.is_empty() {
+            *slot = Some(iroh.clone());
+            return Some(iroh);
+        }
+        drop(slot);
+
+        info!(self, "Closing iroh, no realtime channel uses it.");
+        iroh.close().await.log_err(self).ok();
+        None
     }
 
     pub(crate) async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: NodeAddr) -> Result<()> {
-        if let Some(iroh) = &*self.iroh.read().await {
+        if let Some(iroh) = self.get_active_iroh().await {
             info!(
                 self,
                 "Adding (maybe existing) peer with id {} to {topic}.", peer.node_id
@@ -460,6 +560,10 @@ pub(crate) async fn get_iroh_topic_for_msg(
 
 /// Send a gossip advertisement to the chat that [MsgId] belongs to.
 /// This method should be called from the frontend when `joinRealtimeChannel` is called.
+///
+/// The channel is joined even without a working relay,
+/// but no advertisement is sent then
+/// because peers could not reach us anyway.
 pub async fn send_webxdc_realtime_advertisement(
     ctx: &Context,
     msg_id: MsgId,
@@ -468,8 +572,15 @@ pub async fn send_webxdc_realtime_advertisement(
         return Ok(None);
     }
 
-    let iroh = ctx.get_or_try_init_peer_channel().await?;
+    if let Some(topic) = get_iroh_topic_for_msg(ctx, msg_id).await? {
+        ctx.left_topics.lock().remove(&topic);
+    }
+    let iroh = ctx.get_active_or_init_iroh().await?;
     let conn = iroh.join_and_subscribe_gossip(ctx, msg_id).await?;
+    if iroh.working_relay_url.is_none() {
+        warn!(ctx, "Not sending realtime advertisement without a relay.");
+        return Ok(conn);
+    }
 
     let webxdc = Message::load_from_db(ctx, msg_id).await?;
     let mut msg = Message::new(Viewtype::Text);
@@ -482,12 +593,15 @@ pub async fn send_webxdc_realtime_advertisement(
 }
 
 /// Send realtime data to other peers using iroh.
+///
+/// This works even without a working relay of our own
+/// because joined peers are dialed through their advertised relays.
 pub async fn send_webxdc_realtime_data(ctx: &Context, msg_id: MsgId, data: Vec<u8>) -> Result<()> {
     if !ctx.get_config_bool(Config::WebxdcRealtimeEnabled).await? {
         return Ok(());
     }
 
-    let iroh = ctx.get_or_try_init_peer_channel().await?;
+    let iroh = ctx.get_active_or_init_iroh().await?;
     iroh.send_webxdc_realtime_data(ctx, msg_id, data).await?;
     Ok(())
 }
@@ -498,10 +612,11 @@ pub async fn send_webxdc_realtime_data(ctx: &Context, msg_id: MsgId, data: Vec<u
 /// `send_webxdc_realtime_*()` functions aren't called for the given `msg_id` anymore until the app
 /// is open again.
 pub async fn leave_webxdc_realtime(ctx: &Context, msg_id: MsgId) -> Result<()> {
-    let Some(iroh) = ctx.get_peer_channels().await else {
+    let Some(topic) = get_iroh_topic_for_msg(ctx, msg_id).await? else {
         return Ok(());
     };
-    let Some(topic) = get_iroh_topic_for_msg(ctx, msg_id).await? else {
+    ctx.left_topics.lock().insert(topic);
+    let Some(iroh) = ctx.get_active_iroh().await else {
         return Ok(());
     };
     iroh.leave_realtime(topic).await?;

src/peer_channels/peer_channels_tests.rs

@@ -2,16 +2,267 @@ use super::*;
 use crate::{
     EventType,
     chat::{self, ChatId, add_contact_to_chat, resend_msgs, send_msg},
+    imap::ServerMetadata,
     message::{Message, Viewtype},
     receive_imf::receive_imf,
     test_utils::{TestContext, TestContextManager},
+    transport::add_pseudo_transport,
 };
 
+/// CI chatmail relay is used as a known-working candidate because
+/// mocking out the serving of https requests is not worth it, and,
+/// besides, it's also useful to exercise production code paths
+/// which the core Python tests do a lot already.
+const WORKING_RELAY: &str = "https://ci-chatmail.testrun.org";
+
+/// Sets the iroh relay a transport announces via IMAP METADATA.
+async fn set_iroh_relay(ctx: &TestContext, transport_id: u32, url: &str) -> Result<()> {
+    ctx.metadata.write().await.insert(
+        transport_id,
+        ServerMetadata {
+            iroh_relay: Some(Url::parse(url)?),
+            ..Default::default()
+        },
+    );
+    Ok(())
+}
+
+/// Adds a transport announcing an iroh relay,
+/// like a chatmail server does via IMAP METADATA.
+async fn announce_relay(ctx: &TestContext, addr: &str, url: &str) -> Result<()> {
+    add_pseudo_transport(ctx, addr).await?;
+    let (_, transport_id) = published_transports(ctx)
+        .await?
+        .into_iter()
+        .find(|(a, _)| a == addr)
+        .context("Transport not found")?;
+    set_iroh_relay(ctx, transport_id, url).await
+}
+
+impl TestContext {
+    /// Announces the working relay for the transport this context has.
+    async fn with_working_iroh_relay(self) -> Self {
+        let (_, transport_id) = published_transports(&self)
+            .await
+            .expect("Transports should be readable")
+            .into_iter()
+            .next()
+            .expect("Context should have a published transport");
+        set_iroh_relay(&self, transport_id, WORKING_RELAY)
+            .await
+            .expect("Relay should be announced");
+        self
+    }
+}
+
+#[test]
+fn test_relay_probe_url() {
+    let probe = |url| relay_probe_url(&Url::parse(url).unwrap()).unwrap();
+    assert_eq!(
+        probe("https://relay.example.org").as_str(),
+        "https://relay.example.org/generate_204"
+    );
+    assert_eq!(
+        probe("https://relay.example.org/some/path").as_str(),
+        "https://relay.example.org/some/path/generate_204"
+    );
+}
+
+/// Sends a webxdc instance and returns it.
+async fn send_webxdc(ctx: &TestContext, peer: &TestContext) -> Result<Message> {
+    let chat = ctx.create_chat(peer).await;
+    let mut instance = Message::new(Viewtype::File);
+    instance.set_file_from_bytes(
+        ctx,
+        "minimal.xdc",
+        include_bytes!("../../test-data/webxdc/minimal.xdc"),
+        None,
+    )?;
+    send_msg(ctx, chat.id, &mut instance).await?;
+    let webxdc = ctx.get_last_msg().await;
+    ctx.pop_sent_msg().await;
+    Ok(webxdc)
+}
+
+/// Returns the relay of the node address to advertise, if any.
+async fn selected_iroh_relay(ctx: &TestContext) -> Result<Option<RelayUrl>> {
+    let iroh = ctx.get_active_or_init_iroh().await?;
+    Ok(iroh
+        .get_relay_node_addr()
+        .ok()
+        .and_then(|addr| addr.relay_url().cloned()))
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_select_working_iroh_relay() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice = &mut tcm.alice().await;
+
+    announce_relay(alice, "one@example.net", "https://127.0.0.1:9").await?;
+    assert_eq!(selected_iroh_relay(alice).await?, None);
+    alice.assert_warn("No working iroh relay").await;
+
+    // Relays announced by unpublished transports are not used
+    announce_relay(alice, "two@example.net", WORKING_RELAY).await?;
+    alice
+        .set_transport_unpublished("two@example.net", true)
+        .await?;
+    assert_eq!(selected_iroh_relay(alice).await?, None);
+    alice.assert_warn("No working iroh relay").await;
+
+    // The endpoint is initialized again on the next use
+    // because it has no working relay and is not in use yet.
+    announce_relay(alice, "three@example.net", "https://192.0.2.1").await?;
+    announce_relay(alice, "four@example.net", WORKING_RELAY).await?;
+    announce_relay(alice, "five@example.net", "https://192.0.2.2").await?;
+    assert_eq!(
+        selected_iroh_relay(alice).await?,
+        Some(RelayUrl::from(Url::parse(WORKING_RELAY)?))
+    );
+    Ok(())
+}
+
+/// Iroh is closed once no channel uses it,
+/// but kept while a channel is joined.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_close_unused_iroh() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice = &mut tcm.alice().await.with_working_iroh_relay().await;
+    let bob = &tcm.bob().await;
+    let alice_webxdc = send_webxdc(alice, bob).await?;
+
+    // A joined channel keeps iroh.
+    send_webxdc_realtime_advertisement(alice, alice_webxdc.id).await?;
+    alice.pop_sent_msg().await;
+    alice.close_unused_iroh().await;
+    assert!(alice.iroh.read().await.is_some());
+
+    // Without a channel iroh is closed and initialized again on demand.
+    leave_webxdc_realtime(alice, alice_webxdc.id).await?;
+    alice.close_unused_iroh().await;
+    assert!(alice.iroh.read().await.is_none());
+    assert!(alice.get_active_or_init_iroh().await.is_ok());
+    Ok(())
+}
+
+/// An io stop racing an iroh initialization wins,
+/// so no iroh survives it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_stop_io_while_initializing_iroh() -> Result<()> {
+    let alice = &TestContext::new_alice().await;
+
+    // Hold the mutex so that both tasks below wait for it,
+    // the initialization first and the io stop behind it.
+    let guard = alice.iroh_init_mutex.lock().await;
+    let init_ctx = alice.ctx.clone();
+    let init = tokio::spawn(async move { init_ctx.get_active_or_init_iroh().await });
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    let stop_ctx = alice.ctx.clone();
+    let stop = tokio::spawn(async move { stop_ctx.stop_io().await });
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    drop(guard);
+
+    assert!(init.await?.is_err());
+    stop.await?;
+    assert!(alice.iroh.read().await.is_none());
+    Ok(())
+}
+
+/// A join that was pending while its channel was left does not register it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_leave_cancels_pending_join() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice = &tcm.alice().await.with_working_iroh_relay().await;
+    let bob = &tcm.bob().await;
+    let alice_webxdc = send_webxdc(alice, bob).await?;
+    let topic = get_iroh_topic_for_msg(alice, alice_webxdc.id)
+        .await?
+        .unwrap();
+
+    // Marks the topic although iroh is not even initialized yet.
+    leave_webxdc_realtime(alice, alice_webxdc.id).await?;
+
+    // A join that passed its unmark before the leave must not register the channel.
+    let iroh = alice.get_active_or_init_iroh().await?;
+    assert!(
+        iroh.join_and_subscribe_gossip(alice, alice_webxdc.id)
+            .await
+            .is_err()
+    );
+    assert!(!iroh.iroh_channels.read().await.contains_key(&topic));
+
+    assert!(
+        send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
+            .await?
+            .is_some()
+    );
+    alice.pop_sent_msg().await;
+    assert!(iroh.iroh_channels.read().await.contains_key(&topic));
+    Ok(())
+}
+
+/// Iroh without a working relay joins channels
+/// but sends no advertisement.
+/// It is kept while in use and replaced afterwards.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_relayless_endpoint() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice = &mut tcm.alice().await;
+    let bob = &tcm.bob().await;
+
+    let alice_webxdc = send_webxdc(alice, bob).await?;
+
+    // An instance a concurrent task still references is shared,
+    // not closed underneath them.
+    announce_relay(alice, "broken@example.net", "https://127.0.0.1:9").await?;
+    let held = alice.get_active_or_init_iroh().await?;
+    assert!(Arc::ptr_eq(&held, &alice.get_active_or_init_iroh().await?));
+    drop(held);
+    alice.assert_warn("No working iroh relay").await;
+
+    // Without a working relay the channel is joined
+    // but no advertisement is sent.
+    assert!(
+        send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
+            .await?
+            .is_some()
+    );
+    alice.assert_warn("No working iroh relay").await;
+    alice
+        .assert_warn("Not sending realtime advertisement without a relay")
+        .await;
+    assert_eq!(selected_iroh_relay(alice).await?, None);
+
+    // Iroh is kept while its channel is in use,
+    // so a relay announced later is not picked up yet.
+    let working_relay = Some(RelayUrl::from(Url::parse(WORKING_RELAY)?));
+    announce_relay(alice, "working@example.net", WORKING_RELAY).await?;
+    assert_eq!(selected_iroh_relay(alice).await?, None);
+
+    // The unused iroh is replaced on the next use.
+    leave_webxdc_realtime(alice, alice_webxdc.id).await?;
+    assert_eq!(selected_iroh_relay(alice).await?, working_relay);
+
+    // Disabling realtime does not tear down iroh,
+    // but sending and advertising become no-ops.
+    alice
+        .set_config_bool(Config::WebxdcRealtimeEnabled, false)
+        .await?;
+    send_webxdc_realtime_data(alice, alice_webxdc.id, b"ignored".to_vec()).await?;
+    assert!(
+        send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
+            .await?
+            .is_none()
+    );
+    assert!(alice.iroh.read().await.is_some());
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_can_communicate() {
     let mut tcm = TestContextManager::new();
-    let alice = &mut tcm.alice().await;
-    let bob = &mut tcm.bob().await;
+    let alice = &mut tcm.alice().await.with_working_iroh_relay().await;
+    let bob = &mut tcm.bob().await.with_working_iroh_relay().await;
 
     // Alice sends webxdc to bob
     let alice_chat = alice.create_chat(bob).await;
@@ -61,17 +312,16 @@ async fn test_can_communicate() {
         members,
         vec![
             alice
-                .get_or_try_init_peer_channel()
+                .get_active_or_init_iroh()
                 .await
                 .unwrap()
-                .get_node_addr()
-                .await
+                .get_relay_node_addr()
                 .unwrap()
                 .node_id
         ]
     );
 
-    bob.get_or_try_init_peer_channel()
+    bob.get_active_or_init_iroh()
         .await
         .unwrap()
         .join_and_subscribe_gossip(bob, bob_webxdc.id)
@@ -83,7 +333,7 @@ async fn test_can_communicate() {
 
     // Alice sends ephemeral message
     alice
-        .get_or_try_init_peer_channel()
+        .get_active_or_init_iroh()
         .await
         .unwrap()
         .send_webxdc_realtime_data(alice, alice_webxdc.id, "alice -> bob".as_bytes().to_vec())
@@ -104,7 +354,7 @@ async fn test_can_communicate() {
         }
     }
     // Bob sends ephemeral message
-    bob.get_or_try_init_peer_channel()
+    bob.get_active_or_init_iroh()
         .await
         .unwrap()
         .send_webxdc_realtime_data(bob, bob_webxdc.id, "bob -> alice".as_bytes().to_vec())
@@ -136,17 +386,16 @@ async fn test_can_communicate() {
     assert_eq!(
         members,
         vec![
-            bob.get_or_try_init_peer_channel()
+            bob.get_active_or_init_iroh()
                 .await
                 .unwrap()
-                .get_node_addr()
-                .await
+                .get_relay_node_addr()
                 .unwrap()
                 .node_id
         ]
     );
 
-    bob.get_or_try_init_peer_channel()
+    bob.get_active_or_init_iroh()
         .await
         .unwrap()
         .send_webxdc_realtime_data(bob, bob_webxdc.id, "bob -> alice 2".as_bytes().to_vec())
@@ -177,8 +426,8 @@ async fn test_can_communicate() {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_duplicated_out_of_order_advertisement() -> Result<()> {
     let mut tcm = TestContextManager::new();
-    let alice = &mut tcm.alice().await;
-    let bob = &mut tcm.bob().await;
+    let alice = &mut tcm.alice().await.with_working_iroh_relay().await;
+    let bob = &mut tcm.bob().await.with_working_iroh_relay().await;
 
     let alice_chat = alice.create_chat(bob).await;
     let mut instance = Message::new(Viewtype::File);
@@ -223,11 +472,10 @@ async fn test_duplicated_out_of_order_advertisement() -> Result<()> {
         members,
         vec![
             alice
-                .get_or_try_init_peer_channel()
+                .get_active_or_init_iroh()
                 .await
                 .unwrap()
-                .get_node_addr()
-                .await
+                .get_relay_node_addr()
                 .unwrap()
                 .node_id
         ]
@@ -239,8 +487,8 @@ async fn test_duplicated_out_of_order_advertisement() -> Result<()> {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_can_reconnect() {
     let mut tcm = TestContextManager::new();
-    let alice = &mut tcm.alice().await;
-    let bob = &mut tcm.bob().await;
+    let alice = &mut tcm.alice().await.with_working_iroh_relay().await;
+    let bob = &mut tcm.bob().await.with_working_iroh_relay().await;
 
     assert!(
         alice
@@ -289,17 +537,16 @@ async fn test_can_reconnect() {
         members,
         vec![
             alice
-                .get_or_try_init_peer_channel()
+                .get_active_or_init_iroh()
                 .await
                 .unwrap()
-                .get_node_addr()
-                .await
+                .get_relay_node_addr()
                 .unwrap()
                 .node_id
         ]
     );
 
-    bob.get_or_try_init_peer_channel()
+    bob.get_active_or_init_iroh()
         .await
         .unwrap()
         .join_and_subscribe_gossip(bob, bob_webxdc.id)
@@ -311,7 +558,7 @@ async fn test_can_reconnect() {
 
     // Alice sends ephemeral message
     alice
-        .get_or_try_init_peer_channel()
+        .get_active_or_init_iroh()
         .await
         .unwrap()
         .send_webxdc_realtime_data(alice, alice_webxdc.id, "alice -> bob".as_bytes().to_vec())
@@ -360,7 +607,9 @@ async fn test_can_reconnect() {
     // Check that sequence number is persisted when leaving the channel.
     assert_eq!(bob_sequence_number, bob_sequence_number_after);
 
-    bob.get_or_try_init_peer_channel()
+    // Unmark the leave like a fresh advertisement would.
+    bob.left_topics.lock().remove(&bob_topic);
+    bob.get_active_or_init_iroh()
         .await
         .unwrap()
         .join_and_subscribe_gossip(bob, bob_webxdc.id)
@@ -370,7 +619,7 @@ async fn test_can_reconnect() {
         .await
         .unwrap();
 
-    bob.get_or_try_init_peer_channel()
+    bob.get_active_or_init_iroh()
         .await
         .unwrap()
         .send_webxdc_realtime_data(bob, bob_webxdc.id, "bob -> alice".as_bytes().to_vec())
@@ -430,8 +679,8 @@ async fn test_can_reconnect() {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_parallel_connect() {
     let mut tcm = TestContextManager::new();
-    let alice = &mut tcm.alice().await;
-    let bob = &mut tcm.bob().await;
+    let alice = &mut tcm.alice().await.with_working_iroh_relay().await;
+    let bob = &mut tcm.bob().await.with_working_iroh_relay().await;
 
     let chat = alice.create_chat(bob).await.id;
 
@@ -450,8 +699,8 @@ async fn test_parallel_connect() {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_webxdc_resend() {
     let mut tcm = TestContextManager::new();
-    let alice = &mut tcm.alice().await;
-    let bob = &mut tcm.bob().await;
+    let alice = &mut tcm.alice().await.with_working_iroh_relay().await;
+    let bob = &mut tcm.bob().await.with_working_iroh_relay().await;
     let group = chat::create_group(alice, "group chat").await.unwrap();
 
     // Alice sends webxdc to bob
@@ -472,7 +721,7 @@ async fn test_webxdc_resend() {
     connect_alice_bob(alice, group, &mut instance, bob).await;
 
     // fiona joins late
-    let fiona = &mut tcm.fiona().await;
+    let fiona = &mut tcm.fiona().await.with_working_iroh_relay().await;
 
     add_contact_to_chat(alice, group, alice.add_or_lookup_contact_id(fiona).await)
         .await
@@ -618,7 +867,7 @@ async fn test_peer_channels_disabled() {
 
     // This internal function should return error
     // if accidentally called with the setting disabled.
-    assert!(alice.ctx.get_or_try_init_peer_channel().await.is_err());
+    assert!(alice.ctx.get_active_or_init_iroh().await.is_err());
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]

src/scheduler.rs

@@ -445,6 +445,8 @@ async fn inbox_fetch_idle(ctx: &Context, imap: &mut Imap, mut session: Session)
 
     maybe_add_time_based_warnings(ctx).await;
 
+    ctx.close_unused_iroh().await;
+
     match ctx.get_config_i64(Config::LastHousekeeping).await {
         Ok(last_housekeeping_time) => {
             let next_housekeeping_time =

src/transport.rs

@@ -724,6 +724,19 @@ fn maybe_reelect_local_primary(transaction: &mut rusqlite::Transaction) -> Resul
     Ok(Some(new_addr.clone()))
 }
 
+/// Returns the address and id of each published transport, newest first.
+pub(crate) async fn published_transports(context: &Context) -> Result<Vec<(String, u32)>> {
+    context
+        .sql
+        .query_map_vec(
+            "SELECT addr, id FROM transports WHERE is_published=1
+             ORDER BY add_timestamp DESC, id DESC",
+            (),
+            |row| Ok((row.get(0)?, row.get(1)?)),
+        )
+        .await
+}
+
 /// Adds transport entry to the `transports` table with empty configuration.
 pub(crate) async fn add_pseudo_transport(context: &Context, addr: &str) -> Result<()> {
     context.sql