From 92978ddeb4cc6d529e60814bf3cd3bdc3451463c Mon Sep 17 00:00:00 2001 From: pokefan77 Date: Fri, 3 Jul 2026 21:39:56 +0200 Subject: [PATCH 1/4] libp2p-tor: dial a peer's clearnet address before its onion Add is_onion to wait() and DialRequest, and make release_ready serve the first clearnet waiter before any onion at the same priority, so a reachable clearnet address is not starved by an onion dial that hangs on a dead service. Update the existing test call-sites to the new signature and add tests for clearnet-first and onion-only serving. --- libp2p-tor/src/dial_limiter.rs | 115 ++++++++++++++++++++++++++++----- 1 file changed, 98 insertions(+), 17 deletions(-) diff --git a/libp2p-tor/src/dial_limiter.rs b/libp2p-tor/src/dial_limiter.rs index 0dc2942b7..5e5373d0b 100644 --- a/libp2p-tor/src/dial_limiter.rs +++ b/libp2p-tor/src/dial_limiter.rs @@ -148,11 +148,21 @@ impl TorDialLimiter { /// # Errors /// /// Returns an error if the background limiter task has stopped. - pub async fn wait(&self, peer_id: Option) -> Result { + /// + /// `is_onion` marks whether this dial targets an onion address. Within a + /// priority, clearnet dials are served before onion ones, so a reachable + /// clearnet address is not starved by an onion dial that hangs on a dead + /// service. + pub async fn wait( + &self, + peer_id: Option, + is_onion: bool, + ) -> Result { let (permit_sender, permit_receiver) = oneshot::channel(); let request = DialRequest { peer_id, permit_sender, + is_onion, }; self.request_sender @@ -192,6 +202,9 @@ impl Drop for TorDialPermit { struct DialRequest { peer_id: Option, permit_sender: oneshot::Sender, + /// Whether this dial targets an onion address. Used to serve clearnet dials + /// before onion ones within the same priority. + is_onion: bool, } struct PriorityQueue { @@ -227,7 +240,15 @@ impl PriorityQueue { break; } - let request = self.queue.pop_front().expect("queue to be non-empty"); + // Prefer a clearnet (non-onion) waiter over onion ones at this + // priority: an onion connect can hang on a dead service, so serving + // clearnet first lets a reachable address win the scarce slot. + let idx = self + .queue + .iter() + .position(|request| !request.is_onion) + .unwrap_or(0); + let request = self.queue.remove(idx).expect("index to be in bounds"); let permit = TorDialPermit { priority, release_sender: Some(release_sender.clone()), @@ -429,11 +450,11 @@ mod tests { async fn normal_dials_are_serialized_by_concurrency_and_delay() { let limiter = limiter(TorDialPriorityTracker::default()); - let first = limiter.wait(None).await.unwrap(); + let first = limiter.wait(None, false).await.unwrap(); let second = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(None).await.unwrap() } + async move { limiter.wait(None, false).await.unwrap() } }); // Blocked by the concurrency limit of 1. @@ -459,7 +480,7 @@ mod tests { let spawn_wait = |peer: PeerId| { priority_tracker.mark_high_priority(peer); let limiter = limiter.clone(); - tokio::spawn(async move { limiter.wait(Some(peer)).await.unwrap() }) + tokio::spawn(async move { limiter.wait(Some(peer), false).await.unwrap() }) }; let first = spawn_wait(PeerId::random()); let second = spawn_wait(PeerId::random()); @@ -494,18 +515,18 @@ mod tests { let limiter = limiter(priority_tracker.clone()); // Saturate the normal queue (concurrency 1). - let normal = limiter.wait(None).await.unwrap(); + let normal = limiter.wait(None, false).await.unwrap(); let normal_waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(None).await.unwrap() } + async move { limiter.wait(None, false).await.unwrap() } }); let high_peer = PeerId::random(); priority_tracker.mark_high_priority(high_peer); let high_waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(Some(high_peer)).await.unwrap() } + async move { limiter.wait(Some(high_peer), false).await.unwrap() } }); // The high dial proceeds despite the normal queue being saturated. @@ -546,17 +567,17 @@ mod tests { // Saturate the low queue (concurrency 1). let low_peer = PeerId::random(); priority_tracker.mark_low_priority(low_peer); - let low = limiter.wait(Some(low_peer)).await.unwrap(); + let low = limiter.wait(Some(low_peer), false).await.unwrap(); let low_waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(Some(low_peer)).await.unwrap() } + async move { limiter.wait(Some(low_peer), false).await.unwrap() } }); // A normal dial proceeds despite the low queue being saturated. let normal_waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(None).await.unwrap() } + async move { limiter.wait(None, false).await.unwrap() } }); settle().await; @@ -574,10 +595,10 @@ mod tests { // Occupy the normal slot and queue another normal dial so the low queue // stays gated off. - let normal = limiter.wait(None).await.unwrap(); + let normal = limiter.wait(None, false).await.unwrap(); let normal_waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(None).await.unwrap() } + async move { limiter.wait(None, false).await.unwrap() } }); // A low dial that cannot start while a normal dial is waiting. @@ -585,7 +606,7 @@ mod tests { priority_tracker.mark_low_priority(peer); let waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(Some(peer)).await.unwrap() } + async move { limiter.wait(Some(peer), false).await.unwrap() } }); settle().await; @@ -609,10 +630,10 @@ mod tests { let limiter = limiter(priority_tracker.clone()); // Occupy the normal slot, then queue another so the normal queue is non-empty. - let normal = limiter.wait(None).await.unwrap(); + let normal = limiter.wait(None, false).await.unwrap(); let normal_waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(None).await.unwrap() } + async move { limiter.wait(None, false).await.unwrap() } }); // Held back while a normal dial is waiting, despite a free low slot. @@ -620,7 +641,7 @@ mod tests { priority_tracker.mark_low_priority(low_peer); let low_waiter = tokio::spawn({ let limiter = limiter.clone(); - async move { limiter.wait(Some(low_peer)).await.unwrap() } + async move { limiter.wait(Some(low_peer), false).await.unwrap() } }); settle().await; @@ -637,4 +658,64 @@ mod tests { let _ = normal_waiter.await.unwrap(); let _ = low_waiter.await.unwrap(); } + + #[tokio::test(start_paused = true)] + async fn clearnet_dial_is_served_before_a_held_onion_dial() { + let priority_tracker = TorDialPriorityTracker::default(); + let limiter = limiter(priority_tracker.clone()); + + // A peer at Low priority (single global slot), matching the restart case. + let peer = PeerId::random(); + priority_tracker.mark_low_priority(peer); + + // The onion dial takes the only Low slot and is held, simulating a dead + // onion that never connects. + let onion = limiter.wait(Some(peer), true).await.unwrap(); + + // The same peer's clearnet dial and another onion dial both queue up. + let clearnet_waiter = tokio::spawn({ + let limiter = limiter.clone(); + async move { limiter.wait(Some(peer), false).await.unwrap() } + }); + let onion_waiter = tokio::spawn({ + let limiter = limiter.clone(); + async move { limiter.wait(Some(peer), true).await.unwrap() } + }); + + // Nothing can start while the held onion occupies the slot. + settle().await; + assert!(!clearnet_waiter.is_finished()); + assert!(!onion_waiter.is_finished()); + + // Free the slot; the clearnet dial must be served before the queued onion. + drop(onion); + tokio::time::advance(Duration::from_secs(8)).await; + settle().await; + assert!(clearnet_waiter.is_finished()); + assert!(!onion_waiter.is_finished()); + + let _ = clearnet_waiter.await.unwrap(); + } + + #[tokio::test(start_paused = true)] + async fn onion_only_queue_is_still_served() { + let limiter = limiter(TorDialPriorityTracker::default()); + + // With no clearnet waiter present, onion dials must still be released. + let first = limiter.wait(None, true).await.unwrap(); + + let second = tokio::spawn({ + let limiter = limiter.clone(); + async move { limiter.wait(None, true).await.unwrap() } + }); + + settle().await; + assert!(!second.is_finished()); + + drop(first); + tokio::time::advance(Duration::from_secs(4)).await; + settle().await; + assert!(second.is_finished()); + let _ = second.await.unwrap(); + } } From 3ad0d45752eb55e9a79a06e0fc436429ae453ff9 Mon Sep 17 00:00:00 2001 From: pokefan77 Date: Fri, 3 Jul 2026 21:39:57 +0200 Subject: [PATCH 2/4] libp2p-tor: add DialLimiterTransport that gates + times Tor dials post-permit Add an outer DialLimiterTransport wrapper that acquires the Tor dial permit before polling the inner dial, then bounds it with tokio::time::timeout, so the time spent queued for a slot is excluded from the timeout. Remove the dial limiter from TorTransport (it now just connects) and drop the now-dead DialLimiter error variant. --- libp2p-tor/src/lib.rs | 134 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 111 insertions(+), 23 deletions(-) diff --git a/libp2p-tor/src/lib.rs b/libp2p-tor/src/lib.rs index 539f202ce..72b7ba481 100644 --- a/libp2p-tor/src/lib.rs +++ b/libp2p-tor/src/lib.rs @@ -59,10 +59,12 @@ use futures::{ use libp2p::{ Multiaddr, Transport, TransportError, core::transport::{ListenerId, TransportEvent}, + multiaddr::Protocol, }; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; +use std::time::Duration; use thiserror::Error; use tor_rtcompat::tokio::TokioRustlsRuntime; @@ -140,9 +142,6 @@ pub struct TorTransport { /// The Tor client. client: Arc>, - /// Limiter for outbound Tor dials. - dial_limiter: Option, - /// Onion services we are listening on. #[cfg(feature = "listen-onion-service")] listeners: HashMap, @@ -210,7 +209,6 @@ impl TorTransport { Self { conversion_mode, client, - dial_limiter: None, #[cfg(feature = "listen-onion-service")] listeners: HashMap::new(), #[cfg(feature = "listen-onion-service")] @@ -233,13 +231,6 @@ impl TorTransport { self } - /// Set a shared outbound Tor dial limiter. - #[must_use] - pub fn with_dial_limiter(mut self, dial_limiter: TorDialLimiter) -> Self { - self.dial_limiter = Some(dial_limiter); - self - } - /// Registers an already-launched onion service: applies bounded-concurrency /// rendezvous handling, extracts the multiaddr, and stores the service. /// @@ -345,8 +336,6 @@ impl TorTransport { pub enum TorTransportError { #[error(transparent)] Client(#[from] TorError), - #[error(transparent)] - DialLimiter(#[from] TorDialLimiterError), #[cfg(feature = "listen-onion-service")] #[error(transparent)] Service(#[from] tor_hsservice::ClientError), @@ -470,18 +459,8 @@ impl Transport for TorTransport { let tor_address = maybe_tor_addr.ok_or(TransportError::MultiaddrNotSupported(addr.clone()))?; let onion_client = self.client.clone(); - let dial_limiter = self.dial_limiter.clone(); - let peer_id = extract_peer_id(&addr); Ok(Box::pin(async move { - // Hold the dial permit for the entire duration of the dial: the slot - // is only freed once `_dial_permit` is dropped at the end of this - // scope (whether the connect succeeded or failed). - let _dial_permit = match dial_limiter { - Some(dial_limiter) => Some(dial_limiter.wait(peer_id).await?), - None => None, - }; - let stream = onion_client.connect(tor_address).await?; tracing::debug!(%addr, "Established connection to peer through Tor"); @@ -586,3 +565,112 @@ impl Transport for TorTransport { Poll::Pending } } + +/// Wraps a transport so every outbound dial first waits for a slot from a shared +/// [`TorDialLimiter`], and only then runs the inner dial under a timeout. +/// +/// This is meant to be the **outermost** transport. The limiter permit is +/// acquired *before* the inner `dial()` future is polled, so time spent queued +/// for a Tor slot is excluded from the dial `timeout`. The permit is held until +/// the inner dial (connect and upgrade) completes or the timeout fires, and is +/// released as soon as this dial is dropped (e.g. cancelled once a sibling +/// address connects). +pub struct DialLimiterTransport { + inner: T, + limiter: TorDialLimiter, + timeout: Duration, +} + +impl DialLimiterTransport { + /// Wraps `inner`, gating its dials through `limiter` and bounding each dial + /// (measured *after* the permit is granted) by `timeout`. + #[must_use] + pub fn new(inner: T, limiter: TorDialLimiter, timeout: Duration) -> Self { + Self { + inner, + limiter, + timeout, + } + } + + /// Wraps an already-created inner dial future so that it only runs once a Tor + /// dial permit has been granted, then bounds it by `timeout`. + fn gate( + &self, + addr: &Multiaddr, + inner_dial: T::Dial, + ) -> BoxFuture<'static, Result> + where + T: Transport, + T::Dial: Send + 'static, + T::Output: Send + 'static, + { + let is_onion = addr + .iter() + .any(|protocol| matches!(protocol, Protocol::Onion3(_))); + let peer_id = extract_peer_id(addr); + let limiter = self.limiter.clone(); + let timeout = self.timeout; + + Box::pin(async move { + // Acquiring the permit here, before polling `inner_dial`, is what + // keeps the queue-wait out of the `timeout` below. + let _permit = limiter + .wait(peer_id, is_onion) + .await + .map_err(std::io::Error::other)?; + + tokio::time::timeout(timeout, inner_dial) + .await + .map_err(|_| std::io::Error::from(std::io::ErrorKind::TimedOut))? + }) + } +} + +impl Transport for DialLimiterTransport +where + T: Transport + Unpin, + T::Dial: Send + 'static, + T::Output: Send + 'static, +{ + type Output = T::Output; + type Error = std::io::Error; + type Dial = BoxFuture<'static, Result>; + type ListenerUpgrade = T::ListenerUpgrade; + + fn listen_on( + &mut self, + id: ListenerId, + addr: Multiaddr, + ) -> Result<(), TransportError> { + self.inner.listen_on(id, addr) + } + + fn remove_listener(&mut self, id: ListenerId) -> bool { + self.inner.remove_listener(id) + } + + fn dial(&mut self, addr: Multiaddr) -> Result> { + let inner_dial = self.inner.dial(addr.clone())?; + Ok(self.gate(&addr, inner_dial)) + } + + fn dial_as_listener( + &mut self, + addr: Multiaddr, + ) -> Result> { + let inner_dial = self.inner.dial_as_listener(addr.clone())?; + Ok(self.gate(&addr, inner_dial)) + } + + fn address_translation(&self, listen: &Multiaddr, observed: &Multiaddr) -> Option { + self.inner.address_translation(listen, observed) + } + + fn poll( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll(cx) + } +} From b6ec5ae1aad72f8b48502898f853152bec6a5487 Mon Sep 17 00:00:00 2001 From: pokefan77 Date: Fri, 3 Jul 2026 21:39:57 +0200 Subject: [PATCH 3/4] swap: apply the per-address dial timeout post-permit on the CLI transport Extract authenticate_and_multiplex_no_timeout and keep authenticate_and_multiplex as that plus the timeout so the ASB is unchanged. Rewire the CLI transport to wrap the no-timeout upgrade with DialLimiterTransport when Tor is enabled (timeout measured after the permit) and a plain timeout when Tor is disabled. Add a CHANGELOG entry. --- CHANGELOG.md | 1 + swap/src/cli/transport.rs | 55 +++++++++++++++++++---------------- swap/src/network/transport.rs | 26 +++++++++++++++-- 3 files changed, 54 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 256fddcc1..0e953852e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] - ASB: The Hermes protocol is now enabled by default (`hermes_enabled` defaults to `true`), and the default `hermes_min_swap_amount` was lowered from `0.01` to `0.001` BTC (~50 USD at a reference price of 50,000 USD/BTC). +- GUI: unreachable multiaddress for a maker (e.g. a stale onion/wormhole) no longer blocks that maker's other addresses from being dialed. Each address is dialed with its own timeout that excludes time spent queued for a Tor slot, and reachable clearnet addresses are tried first. ## [4.11.4] - 2026-06-30 diff --git a/swap/src/cli/transport.rs b/swap/src/cli/transport.rs index 6d24da3c7..4c3caa4a6 100644 --- a/swap/src/cli/transport.rs +++ b/swap/src/cli/transport.rs @@ -2,7 +2,7 @@ use std::num::NonZeroUsize; use std::sync::Arc; use std::time::Duration; -use crate::network::transport::authenticate_and_multiplex; +use crate::network::transport::authenticate_and_multiplex_no_timeout; use anyhow::Result; use arti_client::TorClient; use libp2p::core::muxing::StreamMuxerBox; @@ -10,7 +10,8 @@ use libp2p::core::transport::{Boxed, OptionalTransport}; use libp2p::{PeerId, Transport, identity}; use libp2p::{dns, tcp, websocket}; use libp2p_tor::{ - AddressConversion, TorDialLimiter, TorDialPriorityConfig, TorDialPriorityTracker, TorTransport, + AddressConversion, DialLimiterTransport, TorDialLimiter, TorDialPriorityConfig, + TorDialPriorityTracker, TorTransport, }; use tor_rtcompat::tokio::TokioRustlsRuntime; @@ -23,6 +24,11 @@ const TOR_DIAL_NORMAL_PRIORITY_MIN_DELAY: Duration = Duration::from_secs(1); const TOR_DIAL_LOW_PRIORITY_MAX_CONCURRENT: usize = 1; const TOR_DIAL_LOW_PRIORITY_MIN_DELAY: Duration = Duration::from_secs(4); +/// Per-address dial timeout. Applied *after* a Tor dial slot is granted (see +/// [`DialLimiterTransport`]) so time spent queued for a slot is excluded; when +/// Tor is disabled there is no queue, so a plain timeout is equivalent. +const DIAL_TIMEOUT: Duration = Duration::from_secs(15); + fn new_tor_dial_limiter() -> (TorDialLimiter, TorDialPriorityTracker) { let priority_tracker = TorDialPriorityTracker::default(); @@ -76,16 +82,10 @@ pub fn new( let ws_inner_tcp = tcp::tokio::Transport::new(tcp::Config::new().nodelay(true)); let ws_inner_tcp_dns = dns::tokio::Transport::system(ws_inner_tcp)?; let ws_inner_tor: OptionalTransport = match &maybe_tor_client { - Some(client) => { - let mut transport = - TorTransport::from_client(Arc::clone(client), AddressConversion::IpAndDns); - - if let Some(dial_limiter) = maybe_tor_dial_limiter.clone() { - transport = transport.with_dial_limiter(dial_limiter); - } - - OptionalTransport::some(transport) - } + Some(client) => OptionalTransport::some(TorTransport::from_client( + Arc::clone(client), + AddressConversion::IpAndDns, + )), None => OptionalTransport::none(), }; let ws_inner = ws_inner_tor.or_transport(ws_inner_tcp_dns); @@ -95,15 +95,10 @@ pub fn new( let tcp = tcp::tokio::Transport::new(tcp::Config::new().nodelay(true)); let tcp_with_dns = dns::tokio::Transport::system(tcp)?; let maybe_tor_transport: OptionalTransport = match maybe_tor_client { - Some(client) => { - let mut transport = TorTransport::from_client(client, AddressConversion::IpAndDns); - - if let Some(dial_limiter) = maybe_tor_dial_limiter { - transport = transport.with_dial_limiter(dial_limiter); - } - - OptionalTransport::some(transport) - } + Some(client) => OptionalTransport::some(TorTransport::from_client( + client, + AddressConversion::IpAndDns, + )), None => OptionalTransport::none(), }; let plain_transport = maybe_tor_transport.or_transport(tcp_with_dns); @@ -113,8 +108,18 @@ pub fn new( // /ws suffix) and establish a raw connection without a WebSocket handshake. let transport = ws_transport.or_transport(plain_transport).boxed(); - Ok(( - authenticate_and_multiplex(transport, identity)?, - maybe_tor_priority_tracker, - )) + // Apply the noise + yamux upgrade without a timeout here; the timeout is + // applied by `DialLimiterTransport` *after* the Tor dial permit is granted, + // so that time spent queued for a scarce Tor slot is not counted against it. + // When Tor is disabled there is no queue, so a plain timeout is equivalent. + let upgraded = authenticate_and_multiplex_no_timeout(transport, identity)?; + + let transport = match maybe_tor_dial_limiter { + Some(dial_limiter) => { + DialLimiterTransport::new(upgraded, dial_limiter, DIAL_TIMEOUT).boxed() + } + None => upgraded.timeout(DIAL_TIMEOUT).boxed(), + }; + + Ok((transport, maybe_tor_priority_tracker)) } diff --git a/swap/src/network/transport.rs b/swap/src/network/transport.rs index b401e09b0..ed48f4ac4 100644 --- a/swap/src/network/transport.rs +++ b/swap/src/network/transport.rs @@ -12,12 +12,17 @@ const AUTH_AND_MULTIPLEX_TIMEOUT: Duration = Duration::from_secs(15); const MAX_NUM_STREAMS: usize = 5; /// "Completes" a transport by applying the authentication and multiplexing -/// upgrades. +/// upgrades, **without** a dial timeout. +/// +/// Callers that need a timeout should either use [`authenticate_and_multiplex`] +/// (which wraps this with a fixed timeout) or apply their own. This split exists +/// so the CLI can apply the timeout *after* a Tor dial slot has been acquired, +/// keeping the time queued for a slot out of the timeout. /// /// Even though the actual transport technology in use might be different, for /// two libp2p applications to be compatible, the authentication and /// multiplexing upgrades need to be compatible. -pub fn authenticate_and_multiplex( +pub fn authenticate_and_multiplex_no_timeout( transport: Boxed, identity: &identity::Keypair, ) -> Result> @@ -33,9 +38,24 @@ where .upgrade(Version::V1) .authenticate(auth_upgrade) .multiplex(multiplex_upgrade) - .timeout(AUTH_AND_MULTIPLEX_TIMEOUT) .map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer))) .boxed(); Ok(transport) } + +/// "Completes" a transport by applying the authentication and multiplexing +/// upgrades, with a fixed dial timeout applied to the whole dial. +pub fn authenticate_and_multiplex( + transport: Boxed, + identity: &identity::Keypair, +) -> Result> +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + let transport = authenticate_and_multiplex_no_timeout(transport, identity)? + .timeout(AUTH_AND_MULTIPLEX_TIMEOUT) + .boxed(); + + Ok(transport) +} From 7149353ff42323a5727c952672d4937c4ed9e6c9 Mon Sep 17 00:00:00 2001 From: pokefan77 Date: Sat, 4 Jul 2026 10:37:34 +0200 Subject: [PATCH 4/4] swap: apply dial timeout via TransportTimeout on Boxed transport libp2p-core 0.41's Transport trait has no imeout method, so calling .timeout() on a Boxed transport resolved to tokio_stream::StreamExt::timeout and failed to compile. Wrap the boxed transport with TransportTimeout::new(..) instead, in both the CLI transport (non-Tor branch) and authenticate_and_multiplex. --- swap/src/cli/transport.rs | 3 ++- swap/src/network/transport.rs | 9 ++++++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/swap/src/cli/transport.rs b/swap/src/cli/transport.rs index 4c3caa4a6..bebabcc19 100644 --- a/swap/src/cli/transport.rs +++ b/swap/src/cli/transport.rs @@ -6,6 +6,7 @@ use crate::network::transport::authenticate_and_multiplex_no_timeout; use anyhow::Result; use arti_client::TorClient; use libp2p::core::muxing::StreamMuxerBox; +use libp2p::core::transport::timeout::TransportTimeout; use libp2p::core::transport::{Boxed, OptionalTransport}; use libp2p::{PeerId, Transport, identity}; use libp2p::{dns, tcp, websocket}; @@ -118,7 +119,7 @@ pub fn new( Some(dial_limiter) => { DialLimiterTransport::new(upgraded, dial_limiter, DIAL_TIMEOUT).boxed() } - None => upgraded.timeout(DIAL_TIMEOUT).boxed(), + None => TransportTimeout::new(upgraded, DIAL_TIMEOUT).boxed(), }; Ok((transport, maybe_tor_priority_tracker)) diff --git a/swap/src/network/transport.rs b/swap/src/network/transport.rs index ed48f4ac4..3a4f3b205 100644 --- a/swap/src/network/transport.rs +++ b/swap/src/network/transport.rs @@ -2,6 +2,7 @@ use anyhow::Result; use futures::{AsyncRead, AsyncWrite}; use libp2p::core::muxing::StreamMuxerBox; use libp2p::core::transport::Boxed; +use libp2p::core::transport::timeout::TransportTimeout; use libp2p::core::upgrade::Version; use libp2p::noise; use libp2p::{PeerId, Transport, identity, yamux}; @@ -53,9 +54,11 @@ pub fn authenticate_and_multiplex( where T: AsyncRead + AsyncWrite + Unpin + Send + 'static, { - let transport = authenticate_and_multiplex_no_timeout(transport, identity)? - .timeout(AUTH_AND_MULTIPLEX_TIMEOUT) - .boxed(); + let transport = TransportTimeout::new( + authenticate_and_multiplex_no_timeout(transport, identity)?, + AUTH_AND_MULTIPLEX_TIMEOUT, + ) + .boxed(); Ok(transport) }