From d60271d5bf8affa5ecba24a3c5aca8d4231b81ea Mon Sep 17 00:00:00 2001 From: holger krekel Date: Sun, 26 Jul 2026 21:51:19 +0200 Subject: [PATCH 1/7] feat: select a working iroh relay among all candidates from published relays If there are no Iroh relay candidates from the chatmail relays, try nine.testrun.org's iroh relay (for now, soon to be dropped as well) instead of falling back to Iroh's default Number0 relay. Also try to minimize holding locks everywhere. Note that iroh endpoints do not provide any way to determine if the relay is alive and connected if there is no /ping endpoint which currently deployed relays do not offer, so for now and the next couple of months, probing /generate_204 endpoint for http success is the only discriminator i found for a relay working. --- src/config.rs | 17 +-- src/context.rs | 4 + src/mimefactory.rs | 3 +- src/net/http.rs | 39 ++++- src/peer_channels.rs | 177 +++++++++++++++++------ src/peer_channels/peer_channels_tests.rs | 75 +++++++++- src/transport.rs | 13 ++ 7 files changed, 257 insertions(+), 71 deletions(-) diff --git a/src/config.rs b/src/config.rs index 8cf59b025a..e2c2d5d303 100644 --- a/src/config.rs +++ b/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. @@ -952,16 +952,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> { - 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. diff --git a/src/context.rs b/src/context.rs index fc50fa2c06..b16acee093 100644 --- a/src/context.rs +++ b/src/context.rs @@ -317,6 +317,9 @@ pub struct InnerContext { /// Iroh for realtime peer channels. pub(crate) iroh: Arc>>, + /// Mutex to serialize initializations of [`Self::iroh`]. + pub(crate) iroh_init_mutex: Mutex<()>, + /// The own fingerprint, if it was computed already. /// tokio::sync::OnceCell would be possible to use, but overkill for our usecase; /// the standard library's OnceLock is enough, and it's a lot smaller in memory. @@ -506,6 +509,7 @@ impl Context { tls_session_store: TlsSessionStore::new(), spki_hash_store: SpkiHashStore::new(), iroh: Arc::new(RwLock::new(None)), + iroh_init_mutex: Mutex::new(()), self_fingerprint: OnceLock::new(), self_public_key: Mutex::new(None), published_connectivities: parking_lot::Mutex::new(Vec::new()), diff --git a/src/mimefactory.rs b/src/mimefactory.rs index eac134f809..3a875093ee 100644 --- a/src/mimefactory.rs +++ b/src/mimefactory.rs @@ -1916,8 +1916,7 @@ impl MimeFactory { let node_addr = context .get_or_try_init_peer_channel() .await? - .get_node_addr() - .await?; + .get_working_node_addr()?; // We should not send `null` as relay URL // as this is the only way to reach the node. diff --git a/src/net/http.rs b/src/net/http.rs index b1f9807075..37c4960d70 100644 --- a/src/net/http.rs +++ b/src/net/http.rs @@ -52,6 +52,7 @@ async fn get_http_sender( context: &Context, parsed_url: hyper::Uri, strict_tls: bool, + use_proxy: bool, ) -> Result> 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 = match scheme { "http" => { @@ -279,7 +284,7 @@ async fn fetch_url(context: &Context, original_url: &str, strict_tls: bool) -> R .parse::() .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::() + .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(parsed_url) + .header(hyper::header::HOST, authority.as_str()) + .body(http_body_util::Empty::::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")? diff --git a/src/peer_channels.rs b/src/peer_channels.rs index 4e4be8d7be..702c7a80b3 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -26,28 +26,36 @@ use anyhow::{Context as _, Result, anyhow, bail}; use data_encoding::BASE32_NOPAD; use futures_lite::StreamExt; -use iroh::{Endpoint, NodeAddr, NodeId, PublicKey, RelayMode, RelayUrl, SecretKey}; +use iroh::{Endpoint, NodeAddr, NodeId, PublicKey, RelayMap, 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::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(5); + /// Store Iroh peer channels for the context. #[derive(Debug)] pub struct Iroh { @@ -67,6 +75,9 @@ pub struct Iroh { /// /// This is attached to every message to work around `iroh_gossip` deduplication. pub(crate) public_key: PublicKey, + + /// Home relay URL verified to work, None if peers cannot reach us. + working_relay_url: Option, } impl Iroh { @@ -184,16 +195,21 @@ 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 { - let mut addr = self.router.endpoint().node_addr().await?; - addr.direct_addresses = BTreeSet::new(); - debug_assert!(addr.relay_url().is_some()); - Ok(addr) + /// Returns whether the endpoint can still be used. + async fn is_usable(&self) -> bool { + // We don't have a working relay but might still + // have dialed a peer and have active channels. + self.working_relay_url.is_some() || !self.iroh_channels.read().await.is_empty() + } + + /// Returns the iroh [NodeAddr] with the working relay URL + /// and without direct IP addresses. + pub(crate) fn get_working_node_addr(&self) -> Result { + 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 +247,73 @@ impl ChannelState { } } +/// Selects a working iroh relay among the candidates. +async fn select_iroh_relay(context: &Context, candidates: &[Url]) -> Result { + let probes = candidates.iter().cloned().map(|candidate| { + let context = context.clone(); + async move { + let probe_target = candidate.join("/generate_204")?; + 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) +} + +/// Downgrades a write lock on an initialized `iroh` into a read guard on the value. +fn downgrade_iroh_write_lock( + lock: tokio::sync::RwLockWriteGuard<'_, Option>, +) -> Result> { + tokio::sync::RwLockWriteGuard::try_downgrade_map(lock, |opt_iroh| opt_iroh.as_ref()) + .map_err(|_| anyhow!("Downgrade should succeed as the value is `Some`")) +} + impl Context { /// Create iroh endpoint and gossip. async fn init_peer_channels(&self) -> Result { info!(self, "Initializing peer channels."); - 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() + // Iroh relays from unpublished transports are not advertised. + let published = published_transports(self).await?; + let metadata = self.metadata.read().await; + let mut relay_candidates: Vec = Vec::new(); + for (_, transport_id) in published { + if let Some(url) = metadata + .get(&transport_id) + .and_then(|conf| conf.iroh_relay.clone()) + && !relay_candidates.contains(&url) + { + relay_candidates.push(url); + } + } + drop(metadata); + if relay_candidates.is_empty() { + // FIXME: this should be RelayMode::Disabled instead + // once multi-relay usage makes missing iroh relays rare + // and tests can deal with it (best after Iroh 1.0 upgrade?). + warn!(self, "No iroh relay found, using fallback one."); + relay_candidates.push(Url::parse("https://nine.testrun.org")?); + } + let working_relay_url = select_iroh_relay(self, &relay_candidates) .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 + .context("No working iroh relay") + .log_err(self) + .ok() + .map(RelayUrl::from); + let relay_mode = match &working_relay_url { + Some(relay_url) => RelayMode::Custom(RelayMap::from(relay_url.clone())), + None => RelayMode::Disabled, }; + let secret_key = SecretKey::generate(rand_old::rngs::OsRng); + let public_key = secret_key.public(); let endpoint = Box::pin( Endpoint::builder() .tls_x509() // For compatibility with iroh <0.34.0 @@ -283,6 +344,7 @@ impl Context { sequence_numbers: Mutex::new(HashMap::new()), iroh_channels: RwLock::new(HashMap::new()), public_key, + working_relay_url, }) } @@ -303,26 +365,35 @@ impl Context { bail!("Attempt to initialize Iroh when realtime is disabled"); } - if let Some(lock) = self.get_peer_channels().await { - return Ok(lock); + // Return an already usable endpoint under a read lock so that + // concurrent realtime joins/sends do not serialize on the init mutex. + if let Some(iroh) = self.get_peer_channels().await + && iroh.is_usable().await + { + return Ok(iroh); } - let lock = self.iroh.write().await; - match tokio::sync::RwLockWriteGuard::<'_, std::option::Option>::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>::try_downgrade_map( - lock, - |opt_iroh| opt_iroh.as_ref(), - ) - .map_err(|_| anyhow!("Downgrade should succeed as we just stored `Some` value")) - } + let _guard = self.iroh_init_mutex.lock().await; + + // Check again, another task may have initialized in the meantime. + let mut lock = self.iroh.write().await; + if let Some(iroh) = &*lock + && iroh.is_usable().await + { + return downgrade_iroh_write_lock(lock); + } + + // Drop the unused endpoint that has no working relay and no topics. + let stale = lock.take(); + drop(lock); + if let Some(stale) = stale { + stale.close().await.log_err(self).ok(); } + + let iroh = self.init_peer_channels().await?; + let mut lock = self.iroh.write().await; + *lock = Some(iroh); + downgrade_iroh_write_lock(lock) } pub(crate) async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: NodeAddr) -> Result<()> { @@ -460,6 +531,9 @@ 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. +/// +/// No advertisement is sent without a working relay, +/// peers could not reach us anyway. pub async fn send_webxdc_realtime_advertisement( ctx: &Context, msg_id: MsgId, @@ -468,8 +542,17 @@ pub async fn send_webxdc_realtime_advertisement( return Ok(None); } - let iroh = ctx.get_or_try_init_peer_channel().await?; - let conn = iroh.join_and_subscribe_gossip(ctx, msg_id).await?; + // Rendering the message in send_msg() locks `iroh` again, + // so the guard must not be held across it. + let conn = { + let iroh = ctx.get_or_try_init_peer_channel().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); + } + conn + }; let webxdc = Message::load_from_db(ctx, msg_id).await?; let mut msg = Message::new(Viewtype::Text); diff --git a/src/peer_channels/peer_channels_tests.rs b/src/peer_channels/peer_channels_tests.rs index 84a9de7f43..edb07c1624 100644 --- a/src/peer_channels/peer_channels_tests.rs +++ b/src/peer_channels/peer_channels_tests.rs @@ -2,11 +2,74 @@ 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, }; +/// 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")?; + ctx.metadata.write().await.insert( + transport_id, + ServerMetadata { + iroh_relay: Some(Url::parse(url)?), + ..Default::default() + }, + ); + Ok(()) +} + +/// Returns the relay of the node address to advertise, if any. +async fn selected_iroh_relay(ctx: &TestContext) -> Result> { + let iroh = ctx.get_or_try_init_peer_channel().await?; + Ok(iroh + .get_working_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<()> { + // 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"; + + 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); + + // 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); + + // 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(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_can_communicate() { let mut tcm = TestContextManager::new(); @@ -64,8 +127,7 @@ async fn test_can_communicate() { .get_or_try_init_peer_channel() .await .unwrap() - .get_node_addr() - .await + .get_working_node_addr() .unwrap() .node_id ] @@ -139,8 +201,7 @@ async fn test_can_communicate() { bob.get_or_try_init_peer_channel() .await .unwrap() - .get_node_addr() - .await + .get_working_node_addr() .unwrap() .node_id ] @@ -226,8 +287,7 @@ async fn test_duplicated_out_of_order_advertisement() -> Result<()> { .get_or_try_init_peer_channel() .await .unwrap() - .get_node_addr() - .await + .get_working_node_addr() .unwrap() .node_id ] @@ -292,8 +352,7 @@ async fn test_can_reconnect() { .get_or_try_init_peer_channel() .await .unwrap() - .get_node_addr() - .await + .get_working_node_addr() .unwrap() .node_id ] diff --git a/src/transport.rs b/src/transport.rs index 7c77f4da72..c4efeffb13 100644 --- a/src/transport.rs +++ b/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> { + 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 From cb8fe6dacea4ac6e65155ef7544cbed65386a598 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Mon, 10 Aug 2026 22:16:57 +0200 Subject: [PATCH 2/7] fix: append probe path to iroh relay URL Preserves any path of the relay URL so that a relay served under a prefix like https://example.org/iroh is probed correctly. --- src/peer_channels.rs | 16 +++++++++++++++- src/peer_channels/peer_channels_tests.rs | 13 +++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/peer_channels.rs b/src/peer_channels.rs index 702c7a80b3..53ba4650ca 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -247,12 +247,26 @@ 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 { + 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 { let probes = candidates.iter().cloned().map(|candidate| { let context = context.clone(); async move { - let probe_target = candidate.join("/generate_204")?; + let probe_target = relay_probe_url(&candidate)?; timeout( RELAY_PROBE_TIMEOUT, probe_iroh_url(&context, probe_target.as_str()), diff --git a/src/peer_channels/peer_channels_tests.rs b/src/peer_channels/peer_channels_tests.rs index edb07c1624..f2879d820c 100644 --- a/src/peer_channels/peer_channels_tests.rs +++ b/src/peer_channels/peer_channels_tests.rs @@ -28,6 +28,19 @@ async fn announce_relay(ctx: &TestContext, addr: &str, url: &str) -> Result<()> Ok(()) } +#[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" + ); +} + /// Returns the relay of the node address to advertise, if any. async fn selected_iroh_relay(ctx: &TestContext) -> Result> { let iroh = ctx.get_or_try_init_peer_channel().await?; From 8f0c014ae47cb6144776a5929df059ef4b02d873 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Tue, 11 Aug 2026 01:42:47 +0200 Subject: [PATCH 3/7] refactor: rename iroh functions init_peer_channels() -> init_iroh() get_or_try_init_peer_channel() -> get_active_or_init_iroh() get_working_node_addr() -> get_relay_node_addr() --- src/mimefactory.rs | 4 +-- src/peer_channels.rs | 14 ++++----- src/peer_channels/peer_channels_tests.rs | 38 ++++++++++++------------ 3 files changed, 27 insertions(+), 29 deletions(-) diff --git a/src/mimefactory.rs b/src/mimefactory.rs index 3a875093ee..451e5e1114 100644 --- a/src/mimefactory.rs +++ b/src/mimefactory.rs @@ -1914,9 +1914,9 @@ impl MimeFactory { } SystemMessage::IrohNodeAddr => { let node_addr = context - .get_or_try_init_peer_channel() + .get_active_or_init_iroh() .await? - .get_working_node_addr()?; + .get_relay_node_addr()?; // We should not send `null` as relay URL // as this is the only way to reach the node. diff --git a/src/peer_channels.rs b/src/peer_channels.rs index 53ba4650ca..c4a9830149 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -204,7 +204,7 @@ impl Iroh { /// Returns the iroh [NodeAddr] with the working relay URL /// and without direct IP addresses. - pub(crate) fn get_working_node_addr(&self) -> Result { + pub(crate) fn get_relay_node_addr(&self) -> Result { let relay_url = self .working_relay_url .clone() @@ -292,7 +292,7 @@ fn downgrade_iroh_write_lock( impl Context { /// Create iroh endpoint and gossip. - async fn init_peer_channels(&self) -> Result { + async fn init_iroh(&self) -> Result { info!(self, "Initializing peer channels."); // Iroh relays from unpublished transports are not advertised. let published = published_transports(self).await?; @@ -372,9 +372,7 @@ impl Context { } /// Get or initialize the iroh peer channel. - pub async fn get_or_try_init_peer_channel( - &self, - ) -> Result> { + pub async fn get_active_or_init_iroh(&self) -> Result> { if !self.get_config_bool(Config::WebxdcRealtimeEnabled).await? { bail!("Attempt to initialize Iroh when realtime is disabled"); } @@ -404,7 +402,7 @@ impl Context { stale.close().await.log_err(self).ok(); } - let iroh = self.init_peer_channels().await?; + let iroh = self.init_iroh().await?; let mut lock = self.iroh.write().await; *lock = Some(iroh); downgrade_iroh_write_lock(lock) @@ -559,7 +557,7 @@ pub async fn send_webxdc_realtime_advertisement( // Rendering the message in send_msg() locks `iroh` again, // so the guard must not be held across it. let conn = { - let iroh = ctx.get_or_try_init_peer_channel().await?; + 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."); @@ -584,7 +582,7 @@ pub async fn send_webxdc_realtime_data(ctx: &Context, msg_id: MsgId, data: Vec Result> { - let iroh = ctx.get_or_try_init_peer_channel().await?; + let iroh = ctx.get_active_or_init_iroh().await?; Ok(iroh - .get_working_node_addr() + .get_relay_node_addr() .ok() .and_then(|addr| addr.relay_url().cloned())) } @@ -137,16 +137,16 @@ async fn test_can_communicate() { members, vec![ alice - .get_or_try_init_peer_channel() + .get_active_or_init_iroh() .await .unwrap() - .get_working_node_addr() + .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) @@ -158,7 +158,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()) @@ -179,7 +179,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()) @@ -211,16 +211,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_working_node_addr() + .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()) @@ -297,10 +297,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_working_node_addr() + .get_relay_node_addr() .unwrap() .node_id ] @@ -362,16 +362,16 @@ async fn test_can_reconnect() { members, vec![ alice - .get_or_try_init_peer_channel() + .get_active_or_init_iroh() .await .unwrap() - .get_working_node_addr() + .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) @@ -383,7 +383,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()) @@ -432,7 +432,7 @@ 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() + bob.get_active_or_init_iroh() .await .unwrap() .join_and_subscribe_gossip(bob, bob_webxdc.id) @@ -442,7 +442,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()) @@ -690,7 +690,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)] From 712a46c902e78651d9c120cb534ac0811ff1e48b Mon Sep 17 00:00:00 2001 From: holger krekel Date: Tue, 11 Aug 2026 01:43:49 +0200 Subject: [PATCH 4/7] refactor: hand out Arc instead of lock guards Callers get an Arc instead of an RwLock read guard, so no guard is held across await points and the downgrade-map machinery is gone. get_peer_channels() becomes get_active_iroh() and folds in the usability check. Relay candidate collection moves into published_iroh_relays(). --- src/context.rs | 6 +- src/mimefactory.rs | 4 - src/peer_channels.rs | 145 ++++++++++------------- src/peer_channels/peer_channels_tests.rs | 58 +++++++++ 4 files changed, 125 insertions(+), 88 deletions(-) diff --git a/src/context.rs b/src/context.rs index b16acee093..74cb71633d 100644 --- a/src/context.rs +++ b/src/context.rs @@ -315,7 +315,7 @@ pub struct InnerContext { pub(crate) spki_hash_store: SpkiHashStore, /// Iroh for realtime peer channels. - pub(crate) iroh: Arc>>, + pub(crate) iroh: RwLock>>, /// Mutex to serialize initializations of [`Self::iroh`]. pub(crate) iroh_init_mutex: Mutex<()>, @@ -508,7 +508,7 @@ 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(()), self_fingerprint: OnceLock::new(), self_public_key: Mutex::new(None), @@ -565,7 +565,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; diff --git a/src/mimefactory.rs b/src/mimefactory.rs index 451e5e1114..ffc040adf6 100644 --- a/src/mimefactory.rs +++ b/src/mimefactory.rs @@ -1917,10 +1917,6 @@ impl MimeFactory { .get_active_or_init_iroh() .await? .get_relay_node_addr()?; - - // 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()); headers.push(( HeaderDef::IrohNodeAddr.into(), mail_builder::headers::text::Text::new(serde_json::to_string(&node_addr)?) diff --git a/src/peer_channels.rs b/src/peer_channels.rs index c4a9830149..c2641e2920 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -32,6 +32,7 @@ use iroh_gossip::proto::TopicId; use parking_lot::Mutex; use std::collections::HashMap; use std::env; +use std::sync::Arc; use std::time::Duration; use tokio::sync::{RwLock, oneshot}; use tokio::task::JoinHandle; @@ -76,7 +77,7 @@ pub struct Iroh { /// This is attached to every message to work around `iroh_gossip` deduplication. pub(crate) public_key: PublicKey, - /// Home relay URL verified to work, None if peers cannot reach us. + /// Home relay URL probed successfully when Iroh was created. working_relay_url: Option, } @@ -87,7 +88,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") } @@ -195,13 +196,6 @@ impl Iroh { *entry } - /// Returns whether the endpoint can still be used. - async fn is_usable(&self) -> bool { - // We don't have a working relay but might still - // have dialed a peer and have active channels. - self.working_relay_url.is_some() || !self.iroh_channels.read().await.is_empty() - } - /// Returns the iroh [NodeAddr] with the working relay URL /// and without direct IP addresses. pub(crate) fn get_relay_node_addr(&self) -> Result { @@ -282,36 +276,40 @@ async fn select_iroh_relay(context: &Context, candidates: &[Url]) -> Result Ok(selected) } -/// Downgrades a write lock on an initialized `iroh` into a read guard on the value. -fn downgrade_iroh_write_lock( - lock: tokio::sync::RwLockWriteGuard<'_, Option>, -) -> Result> { - tokio::sync::RwLockWriteGuard::try_downgrade_map(lock, |opt_iroh| opt_iroh.as_ref()) - .map_err(|_| anyhow!("Downgrade should succeed as the value is `Some`")) +/// Returns the deduplicated iroh relay URLs announced by the published transports. +async fn published_iroh_relays(context: &Context) -> Result> { + let published = published_transports(context).await?; + let metadata = context.metadata.read().await; + let mut relays: Vec = 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. + /// 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 { - info!(self, "Initializing peer channels."); - // Iroh relays from unpublished transports are not advertised. - let published = published_transports(self).await?; - let metadata = self.metadata.read().await; - let mut relay_candidates: Vec = Vec::new(); - for (_, transport_id) in published { - if let Some(url) = metadata - .get(&transport_id) - .and_then(|conf| conf.iroh_relay.clone()) - && !relay_candidates.contains(&url) - { - relay_candidates.push(url); - } + if !self.get_config_bool(Config::WebxdcRealtimeEnabled).await? { + bail!("Attempt to initialize Iroh when realtime is disabled"); } - drop(metadata); + info!(self, "Initializing Iroh for realtime channels."); + let mut relay_candidates = published_iroh_relays(self).await?; if relay_candidates.is_empty() { - // FIXME: this should be RelayMode::Disabled instead + // FIXME: this should fail to setup Iroh // once multi-relay usage makes missing iroh relays rare - // and tests can deal with it (best after Iroh 1.0 upgrade?). + // and tests can deal with it (maybe better after Iroh 1.0 upgrade). warn!(self, "No iroh relay found, using fallback one."); relay_candidates.push(Url::parse("https://nine.testrun.org")?); } @@ -325,7 +323,6 @@ impl Context { Some(relay_url) => RelayMode::Custom(RelayMap::from(relay_url.clone())), None => RelayMode::Disabled, }; - let secret_key = SecretKey::generate(rand_old::rngs::OsRng); let public_key = secret_key.public(); let endpoint = Box::pin( @@ -362,54 +359,41 @@ impl Context { }) } - /// Returns [`None`] if the peer channels has not been initialized. - pub async fn get_peer_channels(&self) -> Option> { - tokio::sync::RwLockReadGuard::<'_, std::option::Option>::try_map( - self.iroh.read().await, - |opt_iroh| opt_iroh.as_ref(), - ) - .ok() - } - - /// Get or initialize the iroh peer channel. - pub async fn get_active_or_init_iroh(&self) -> Result> { - if !self.get_config_bool(Config::WebxdcRealtimeEnabled).await? { - bail!("Attempt to initialize Iroh when realtime is disabled"); + /// Returns iroh while it has a working relay + /// or channels through which peers may have been dialed. + pub async fn get_active_iroh(&self) -> Option> { + 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 } + } - // Return an already usable endpoint under a read lock so that - // concurrent realtime joins/sends do not serialize on the init mutex. - if let Some(iroh) = self.get_peer_channels().await - && iroh.is_usable().await - { + /// Returns active iroh, initializing it if necessary. + /// + /// Inactive iroh is replaced, probing the relay candidates again. + pub async fn get_active_or_init_iroh(&self) -> Result> { + if let Some(iroh) = self.get_active_iroh().await { return Ok(iroh); } let _guard = self.iroh_init_mutex.lock().await; - // Check again, another task may have initialized in the meantime. - let mut lock = self.iroh.write().await; - if let Some(iroh) = &*lock - && iroh.is_usable().await - { - return downgrade_iroh_write_lock(lock); + if let Some(iroh) = self.get_active_iroh().await { + return Ok(iroh); } - - // Drop the unused endpoint that has no working relay and no topics. - let stale = lock.take(); - drop(lock); - if let Some(stale) = stale { + if let Some(stale) = self.iroh.write().await.take() { stale.close().await.log_err(self).ok(); } - let iroh = self.init_iroh().await?; - let mut lock = self.iroh.write().await; - *lock = Some(iroh); - downgrade_iroh_write_lock(lock) + let iroh = Arc::new(self.init_iroh().await?); + *self.iroh.write().await = Some(iroh.clone()); + Ok(iroh) } 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 @@ -544,8 +528,9 @@ 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. /// -/// No advertisement is sent without a working relay, -/// peers could not reach us anyway. +/// 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, @@ -554,17 +539,12 @@ pub async fn send_webxdc_realtime_advertisement( return Ok(None); } - // Rendering the message in send_msg() locks `iroh` again, - // so the guard must not be held across it. - let conn = { - 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); - } - conn - }; + 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); @@ -577,6 +557,9 @@ 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) -> Result<()> { if !ctx.get_config_bool(Config::WebxdcRealtimeEnabled).await? { return Ok(()); @@ -593,7 +576,7 @@ pub async fn send_webxdc_realtime_data(ctx: &Context, msg_id: MsgId, data: Vec Result<()> { - let Some(iroh) = ctx.get_peer_channels().await else { + let Some(iroh) = ctx.get_active_iroh().await else { return Ok(()); }; let Some(topic) = get_iroh_topic_for_msg(ctx, msg_id).await? else { diff --git a/src/peer_channels/peer_channels_tests.rs b/src/peer_channels/peer_channels_tests.rs index cc6f9d5f1a..4044c68b86 100644 --- a/src/peer_channels/peer_channels_tests.rs +++ b/src/peer_channels/peer_channels_tests.rs @@ -83,6 +83,64 @@ async fn test_select_working_iroh_relay() -> Result<()> { 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<()> { + const WORKING_RELAY: &str = "https://ci-chatmail.testrun.org"; + + let mut tcm = TestContextManager::new(); + let alice = &mut tcm.alice().await; + let bob = &tcm.bob().await; + + let alice_chat = alice.create_chat(bob).await; + let mut instance = Message::new(Viewtype::File); + instance.set_file_from_bytes( + alice, + "minimal.xdc", + include_bytes!("../../test-data/webxdc/minimal.xdc"), + None, + )?; + send_msg(alice, alice_chat.id, &mut instance).await?; + let alice_webxdc = alice.get_last_msg().await; + alice.pop_sent_msg().await; + + // Without a working relay the channel is joined + // but no advertisement is sent. + announce_relay(alice, "broken@example.net", "https://127.0.0.1:9").await?; + assert!( + send_webxdc_realtime_advertisement(alice, alice_webxdc.id) + .await? + .is_some() + ); + 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(); From 5d0a24394dd2909d4d6d032fb9ffb4ee620bbcd6 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Tue, 11 Aug 2026 01:44:54 +0200 Subject: [PATCH 5/7] fix: do not leave iroh running after stop_io stop_io() now causes a concurrent initialization to be dropped by checking both on the init mutex and detecting if stop_io was called. --- src/context.rs | 10 ++++++++-- src/peer_channels.rs | 8 ++++++++ src/peer_channels/peer_channels_tests.rs | 23 +++++++++++++++++++++++ 3 files changed, 39 insertions(+), 2 deletions(-) diff --git a/src/context.rs b/src/context.rs index 74cb71633d..2b3a0a9363 100644 --- a/src/context.rs +++ b/src/context.rs @@ -4,7 +4,7 @@ use std::collections::{BTreeMap, HashMap}; 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; @@ -317,9 +317,12 @@ pub struct InnerContext { /// Iroh for realtime peer channels. pub(crate) iroh: RwLock>>, - /// Mutex to serialize initializations of [`Self::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, + /// The own fingerprint, if it was computed already. /// tokio::sync::OnceCell would be possible to use, but overkill for our usecase; /// the standard library's OnceLock is enough, and it's a lot smaller in memory. @@ -510,6 +513,7 @@ impl Context { spki_hash_store: SpkiHashStore::new(), iroh: RwLock::new(None), iroh_init_mutex: Mutex::new(()), + io_stop_count: AtomicUsize::new(0), self_fingerprint: OnceLock::new(), self_public_key: Mutex::new(None), published_connectivities: parking_lot::Mutex::new(Vec::new()), @@ -541,7 +545,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. diff --git a/src/peer_channels.rs b/src/peer_channels.rs index c2641e2920..e9faee9420 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -33,6 +33,7 @@ use parking_lot::Mutex; 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; @@ -372,17 +373,24 @@ impl Context { /// 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 async fn get_active_or_init_iroh(&self) -> Result> { if let Some(iroh) = self.get_active_iroh().await { return Ok(iroh); } + 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"); + } + if let Some(stale) = self.iroh.write().await.take() { stale.close().await.log_err(self).ok(); } diff --git a/src/peer_channels/peer_channels_tests.rs b/src/peer_channels/peer_channels_tests.rs index 4044c68b86..9e3f2b23bb 100644 --- a/src/peer_channels/peer_channels_tests.rs +++ b/src/peer_channels/peer_channels_tests.rs @@ -83,6 +83,29 @@ async fn test_select_working_iroh_relay() -> Result<()> { 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(()) +} + /// Iroh without a working relay joins channels /// but sends no advertisement. /// It is kept while in use and replaced afterwards. From 9d7f3ccf3534ecfb984092fdeb34a9392f4e32a3 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Tue, 11 Aug 2026 01:45:38 +0200 Subject: [PATCH 6/7] feat!: require transports to announce an iroh relay Drop the hardcoded nine.testrun.org fallback: accounts whose transports announce no iroh relay get an endpoint without a home relay. Peers cannot reach them but they can still dial peers. --- src/peer_channels.rs | 11 ++-- src/peer_channels/peer_channels_tests.rs | 73 +++++++++++++++--------- 2 files changed, 51 insertions(+), 33 deletions(-) diff --git a/src/peer_channels.rs b/src/peer_channels.rs index e9faee9420..2e7a52c59e 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -306,13 +306,12 @@ impl Context { bail!("Attempt to initialize Iroh when realtime is disabled"); } info!(self, "Initializing Iroh for realtime channels."); - let mut relay_candidates = published_iroh_relays(self).await?; + let relay_candidates = published_iroh_relays(self).await?; if relay_candidates.is_empty() { - // FIXME: this should fail to setup Iroh - // once multi-relay usage makes missing iroh relays rare - // and tests can deal with it (maybe better after Iroh 1.0 upgrade). - warn!(self, "No iroh relay found, using fallback one."); - relay_candidates.push(Url::parse("https://nine.testrun.org")?); + warn!( + self, + "No transport announces an iroh relay, peers cannot reach us." + ); } let working_relay_url = select_iroh_relay(self, &relay_candidates) .await diff --git a/src/peer_channels/peer_channels_tests.rs b/src/peer_channels/peer_channels_tests.rs index 9e3f2b23bb..4dba3bd75e 100644 --- a/src/peer_channels/peer_channels_tests.rs +++ b/src/peer_channels/peer_channels_tests.rs @@ -9,6 +9,24 @@ use crate::{ 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<()> { @@ -18,14 +36,23 @@ async fn announce_relay(ctx: &TestContext, addr: &str, url: &str) -> Result<()> .into_iter() .find(|(a, _)| a == addr) .context("Transport not found")?; - ctx.metadata.write().await.insert( - transport_id, - ServerMetadata { - iroh_relay: Some(Url::parse(url)?), - ..Default::default() - }, - ); - Ok(()) + 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] @@ -52,12 +79,6 @@ async fn selected_iroh_relay(ctx: &TestContext) -> Result> { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_select_working_iroh_relay() -> Result<()> { - // 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"; - let mut tcm = TestContextManager::new(); let alice = &mut tcm.alice().await; @@ -111,8 +132,6 @@ async fn test_stop_io_while_initializing_iroh() -> Result<()> { /// It is kept while in use and replaced afterwards. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_relayless_endpoint() -> Result<()> { - const WORKING_RELAY: &str = "https://ci-chatmail.testrun.org"; - let mut tcm = TestContextManager::new(); let alice = &mut tcm.alice().await; let bob = &tcm.bob().await; @@ -167,8 +186,8 @@ async fn test_relayless_endpoint() -> Result<()> { #[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; @@ -332,8 +351,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); @@ -393,8 +412,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 @@ -583,8 +602,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; @@ -603,8 +622,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 @@ -625,7 +644,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 From d26a08092dd7a6959701261ba07174d1e7630e89 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Tue, 11 Aug 2026 03:00:53 +0200 Subject: [PATCH 7/7] fix: close iroh relays if we likely have network but there are no live channels addresses #7058 by stopping to ping the iroh relay when all realtime channels are left or closed. --- src/net/http.rs | 2 +- src/peer_channels.rs | 58 ++++++++++++++------- src/peer_channels/peer_channels_tests.rs | 66 +++++++++++++++++++----- src/scheduler.rs | 2 + 4 files changed, 97 insertions(+), 31 deletions(-) diff --git a/src/net/http.rs b/src/net/http.rs index 37c4960d70..057ef54a97 100644 --- a/src/net/http.rs +++ b/src/net/http.rs @@ -419,7 +419,7 @@ pub(crate) async fn probe_iroh_url(context: &Context, url: &str) -> Result<()> { .authority() .context("URL has no authority")? .clone(); - let req = hyper::Request::get(parsed_url) + let req = hyper::Request::get(origin_form(&parsed_url)) .header(hyper::header::HOST, authority.as_str()) .body(http_body_util::Empty::::new())?; diff --git a/src/peer_channels.rs b/src/peer_channels.rs index 2e7a52c59e..2450f80582 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -26,7 +26,7 @@ use anyhow::{Context as _, Result, anyhow, bail}; use data_encoding::BASE32_NOPAD; use futures_lite::StreamExt; -use iroh::{Endpoint, NodeAddr, NodeId, PublicKey, RelayMap, RelayMode, RelayUrl, SecretKey}; +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; @@ -307,20 +307,19 @@ impl Context { } info!(self, "Initializing Iroh for realtime channels."); let relay_candidates = published_iroh_relays(self).await?; - if relay_candidates.is_empty() { - warn!( - self, - "No transport announces an iroh relay, peers cannot reach us." - ); - } - let working_relay_url = select_iroh_relay(self, &relay_candidates) - .await - .context("No working iroh relay") - .log_err(self) - .ok() - .map(RelayUrl::from); + 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(RelayMap::from(relay_url.clone())), + Some(relay_url) => RelayMode::Custom(relay_url.clone().into()), None => RelayMode::Disabled, }; let secret_key = SecretKey::generate(rand_old::rngs::OsRng); @@ -361,7 +360,7 @@ impl Context { /// Returns iroh while it has a working relay /// or channels through which peers may have been dialed. - pub async fn get_active_iroh(&self) -> Option> { + async fn get_active_iroh(&self) -> Option> { let iroh = self.iroh.read().await.clone()?; if iroh.working_relay_url.is_some() || !iroh.iroh_channels.read().await.is_empty() { Some(iroh) @@ -375,7 +374,7 @@ impl Context { /// 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 async fn get_active_or_init_iroh(&self) -> Result> { + pub(crate) async fn get_active_or_init_iroh(&self) -> Result> { if let Some(iroh) = self.get_active_iroh().await { return Ok(iroh); } @@ -390,8 +389,10 @@ impl Context { bail!("Io was stopped"); } - if let Some(stale) = self.iroh.write().await.take() { - stale.close().await.log_err(self).ok(); + // 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?); @@ -399,6 +400,27 @@ impl Context { 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> { + 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.get_active_iroh().await { info!( diff --git a/src/peer_channels/peer_channels_tests.rs b/src/peer_channels/peer_channels_tests.rs index 4dba3bd75e..1f7692d6f4 100644 --- a/src/peer_channels/peer_channels_tests.rs +++ b/src/peer_channels/peer_channels_tests.rs @@ -68,6 +68,22 @@ fn test_relay_probe_url() { ); } +/// Sends a webxdc instance and returns it. +async fn send_webxdc(ctx: &TestContext, peer: &TestContext) -> Result { + 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> { let iroh = ctx.get_active_or_init_iroh().await?; @@ -84,6 +100,7 @@ async fn test_select_working_iroh_relay() -> Result<()> { 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?; @@ -91,6 +108,7 @@ async fn test_select_working_iroh_relay() -> Result<()> { .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. @@ -104,6 +122,29 @@ async fn test_select_working_iroh_relay() -> Result<()> { 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)] @@ -136,26 +177,27 @@ async fn test_relayless_endpoint() -> Result<()> { let alice = &mut tcm.alice().await; let bob = &tcm.bob().await; - let alice_chat = alice.create_chat(bob).await; - let mut instance = Message::new(Viewtype::File); - instance.set_file_from_bytes( - alice, - "minimal.xdc", - include_bytes!("../../test-data/webxdc/minimal.xdc"), - None, - )?; - send_msg(alice, alice_chat.id, &mut instance).await?; - let alice_webxdc = alice.get_last_msg().await; - alice.pop_sent_msg().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. - announce_relay(alice, "broken@example.net", "https://127.0.0.1:9").await?; 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, diff --git a/src/scheduler.rs b/src/scheduler.rs index 30a27eb8de..883767dd82 100644 --- a/src/scheduler.rs +++ b/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 =