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..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; @@ -315,7 +315,13 @@ 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 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; @@ -505,7 +511,9 @@ 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), self_fingerprint: OnceLock::new(), self_public_key: Mutex::new(None), published_connectivities: parking_lot::Mutex::new(Vec::new()), @@ -537,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. @@ -561,7 +571,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 eac134f809..ffc040adf6 100644 --- a/src/mimefactory.rs +++ b/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)?) diff --git a/src/net/http.rs b/src/net/http.rs index b1f9807075..057ef54a97 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(origin_form(&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..2450f80582 100644 --- a/src/peer_channels.rs +++ b/src/peer_channels.rs @@ -30,24 +30,34 @@ 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(5); + /// 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, } 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") } @@ -184,16 +197,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 { - 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 { + 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 +242,88 @@ impl ChannelState { } } -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(); +/// 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) +} - let relay_mode = if let Some(relay_url) = self - .metadata - .read() +/// 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 = relay_probe_url(&candidate)?; + timeout( + RELAY_PROBE_TIMEOUT, + probe_iroh_url(&context, probe_target.as_str()), + ) .await - .values() - .next() + .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> { + 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) { - RelayMode::Custom(RelayUrl::from(relay_url).into()) + relays.push(url); + } + } + Ok(relays) +} + +impl Context { + /// 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 { + 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 { - // FIXME: this should be RelayMode::Disabled instead. - // Currently using default relays because otherwise Rust tests fail. - RelayMode::Default + 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 endpoint = Box::pin( Endpoint::builder() .tls_x509() // For compatibility with iroh <0.34.0 @@ -283,50 +354,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::<'_, std::option::Option>::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> { + 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> { - 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> { + 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(lock) = self.get_peer_channels().await { - return Ok(lock); + // 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 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 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> { + 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 +556,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 +568,12 @@ pub async fn send_webxdc_realtime_advertisement( return Ok(None); } - 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."); + return Ok(conn); + } let webxdc = Message::load_from_db(ctx, msg_id).await?; let mut msg = Message::new(Viewtype::Text); @@ -482,12 +586,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) -> 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,7 +605,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 84a9de7f43..1f7692d6f4 100644 --- a/src/peer_channels/peer_channels_tests.rs +++ b/src/peer_channels/peer_channels_tests.rs @@ -2,16 +2,234 @@ 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 { + 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?; + 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_can_communicate() { +async fn test_select_working_iroh_relay() -> Result<()> { let mut tcm = TestContextManager::new(); let alice = &mut tcm.alice().await; - let bob = &mut tcm.bob().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(()) +} + +/// 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.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 +279,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 +300,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 +321,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 +353,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 +393,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 +439,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 +454,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 +504,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 +525,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 +574,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) @@ -370,7 +584,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 +644,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 +664,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 +686,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 +832,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)] 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 = 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