fix: parallel proxy dialing

src-tauri/src/dialer.rs

@@ -0,0 +1,720 @@
+//! Outbound TCP dials of the proxy worker: the destination of a direct route,
+//! and the upstream proxy itself.
+//!
+//! One name lookup per host runs at a time, and a good answer is used again
+//! for a short time. The connection then races across the addresses
+//! (RFC 8305): the first address starts at once, the next one when an attempt
+//! fails or `ATTEMPT_DELAY` passes, and the first connection wins. An address
+//! that never answers (often an IPv6 route that drops packets) costs one
+//! delay, not the whole time limit. A failure is never remembered: the next
+//! dial of the same host starts again.
+
+use futures_util::future::BoxFuture;
+use futures_util::stream::{FuturesUnordered, StreamExt};
+use std::collections::{HashMap, VecDeque};
+use std::future::Future;
+use std::io;
+use std::net::{IpAddr, SocketAddr};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError};
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::sync::watch;
+use tokio::time::Instant;
+
+/// Wait before the next address starts while earlier attempts still run.
+const ATTEMPT_DELAY: Duration = Duration::from_millis(250);
+/// Limit for one attempt while more addresses wait. The last attempt runs
+/// until the limit of the dial.
+const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(4);
+/// Attempts of one dial that run at the same time. Keeps a dial inside the
+/// descriptor budget of one connection.
+const MAX_IN_FLIGHT: usize = 4;
+/// Limit for a caller that waits for a name lookup.
+pub(crate) const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5);
+/// Until a lookup has run this long, every caller for the same host waits for
+/// it. A stuck system resolver then gets one blocking call per host, not one
+/// per connection, and each caller still waits for an answer that can come.
+const STALE_LOOKUP: Duration = LOOKUP_TIMEOUT;
+/// How long a good answer is used again.
+const ANSWER_TTL: Duration = Duration::from_secs(30);
+const MAX_HOSTS: usize = 2048;
+
+type Answer = Result<Arc<[SocketAddr]>, (io::ErrorKind, String)>;
+pub(crate) type Lookup =
+  Arc<dyn Fn(String) -> BoxFuture<'static, io::Result<Vec<SocketAddr>>> + Send + Sync>;
+
+enum Entry {
+  Ready {
+    addrs: Arc<[SocketAddr]>,
+    until: Instant,
+  },
+  Pending {
+    id: u64,
+    started: Instant,
+    done: watch::Receiver<Option<Answer>>,
+  },
+}
+
+struct ResolverInner {
+  lookup: Lookup,
+  hosts: Mutex<HashMap<String, Entry>>,
+  next_id: AtomicU64,
+}
+
+/// Name lookups shared by every dial of the worker. Cheap to clone.
+#[derive(Clone)]
+pub(crate) struct HostResolver {
+  inner: Arc<ResolverInner>,
+}
+
+fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
+  mutex.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
+impl HostResolver {
+  pub(crate) fn new(lookup: Lookup) -> Self {
+    Self {
+      inner: Arc::new(ResolverInner {
+        lookup,
+        hosts: Mutex::new(HashMap::new()),
+        next_id: AtomicU64::new(0),
+      }),
+    }
+  }
+
+  /// The system resolver (`getaddrinfo`).
+  pub(crate) fn system() -> &'static HostResolver {
+    static SYSTEM: OnceLock<HostResolver> = OnceLock::new();
+    SYSTEM.get_or_init(|| {
+      HostResolver::new(Arc::new(
+        |host: String| -> BoxFuture<'static, io::Result<Vec<SocketAddr>>> {
+          Box::pin(async move { Ok(tokio::net::lookup_host((host.as_str(), 0)).await?.collect()) })
+        },
+      ))
+    })
+  }
+
+  /// Addresses of `host` with port 0, in the order of the resolver.
+  pub(crate) async fn resolve(&self, host: &str, limit: Duration) -> io::Result<Arc<[SocketAddr]>> {
+    let key = host.to_ascii_lowercase();
+    let mut done = {
+      let mut hosts = lock(&self.inner.hosts);
+      let now = Instant::now();
+      match hosts.get(&key) {
+        Some(Entry::Ready { addrs, until }) if *until > now => return Ok(addrs.clone()),
+        Some(Entry::Pending { started, done, .. })
+          if now.duration_since(*started) < STALE_LOOKUP =>
+        {
+          done.clone()
+        }
+        _ => self.start_lookup(&mut hosts, key),
+      }
+    };
+    let limit = limit.min(LOOKUP_TIMEOUT);
+    let answer = match tokio::time::timeout(limit, done.wait_for(Option::is_some)).await {
+      Ok(Ok(answer)) => (*answer).clone(),
+      Ok(Err(_)) => None,
+      Err(_) => {
+        return Err(io::Error::new(
+          io::ErrorKind::TimedOut,
+          format!("name lookup for {host} timed out after {limit:?}"),
+        ))
+      }
+    };
+    match answer {
+      Some(Ok(addrs)) => Ok(addrs),
+      Some(Err((kind, e))) => Err(io::Error::new(
+        kind,
+        format!("name lookup for {host} failed: {e}"),
+      )),
+      None => Err(io::Error::other(format!("name lookup for {host} stopped"))),
+    }
+  }
+
+  /// The lookup runs in its own task, so it ends and fills the cache even when
+  /// every caller has stopped waiting.
+  fn start_lookup(
+    &self,
+    hosts: &mut HashMap<String, Entry>,
+    key: String,
+  ) -> watch::Receiver<Option<Answer>> {
+    if hosts.len() >= MAX_HOSTS {
+      let now = Instant::now();
+      hosts.retain(|_, entry| match entry {
+        Entry::Ready { until, .. } => *until > now,
+        Entry::Pending { .. } => true,
+      });
+      if hosts.len() >= MAX_HOSTS {
+        hosts.retain(|_, entry| matches!(entry, Entry::Pending { .. }));
+      }
+    }
+    let (tx, rx) = watch::channel(None);
+    let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
+    hosts.insert(
+      key.clone(),
+      Entry::Pending {
+        id,
+        started: Instant::now(),
+        done: rx.clone(),
+      },
+    );
+    let lookup = (self.inner.lookup)(key.clone());
+    let inner = Arc::clone(&self.inner);
+    tokio::spawn(async move {
+      let answer: Answer = match lookup.await {
+        Ok(found) => {
+          let mut addrs: Vec<SocketAddr> = Vec::with_capacity(found.len());
+          for addr in found {
+            if !addrs.contains(&addr) {
+              addrs.push(addr);
+            }
+          }
+          if addrs.is_empty() {
+            Err((io::ErrorKind::NotFound, "no address".to_string()))
+          } else {
+            Ok(addrs.into())
+          }
+        }
+        Err(e) => Err((e.kind(), e.to_string())),
+      };
+      inner.finish(&key, id, &answer);
+      let _ = tx.send(Some(answer));
+    });
+    rx
+  }
+
+  /// Put `winner` first for the next dials of `host`, so an address that
+  /// does not answer costs the delay once per answer, not once per dial.
+  pub(crate) fn prefer(&self, host: &str, winner: IpAddr) {
+    let mut hosts = lock(&self.inner.hosts);
+    let Some(Entry::Ready { addrs, .. }) = hosts.get_mut(&host.to_ascii_lowercase()) else {
+      return;
+    };
+    if let Some(position) = addrs.iter().position(|a| a.ip() == winner) {
+      if position > 0 {
+        let mut reordered = addrs.to_vec();
+        let first = reordered.remove(position);
+        reordered.insert(0, first);
+        *addrs = reordered.into();
+      }
+    }
+  }
+}
+
+impl ResolverInner {
+  /// A good answer is kept; a failed one is dropped, so the next caller looks
+  /// the name up again.
+  fn finish(&self, key: &str, id: u64, answer: &Answer) {
+    let mut hosts = lock(&self.hosts);
+    match answer {
+      Ok(addrs) => {
+        hosts.insert(
+          key.to_string(),
+          Entry::Ready {
+            addrs: addrs.clone(),
+            until: Instant::now() + ANSWER_TTL,
+          },
+        );
+      }
+      Err(_) => {
+        if matches!(hosts.get(key), Some(Entry::Pending { id: pending, .. }) if *pending == id) {
+          hosts.remove(key);
+        }
+      }
+    }
+  }
+}
+
+/// Address families of the attempts of one dial.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(crate) struct Families {
+  ipv4: bool,
+  ipv6: bool,
+}
+
+impl Families {
+  fn add(&mut self, addr: SocketAddr) {
+    if addr.is_ipv4() {
+      self.ipv4 = true;
+    } else {
+      self.ipv6 = true;
+    }
+  }
+}
+
+impl std::fmt::Display for Families {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    f.write_str(match (self.ipv4, self.ipv6) {
+      (true, true) => "ipv4,ipv6",
+      (true, false) => "ipv4",
+      (false, true) => "ipv6",
+      (false, false) => "none",
+    })
+  }
+}
+
+/// Timings of a race across addresses.
+pub(crate) struct Race {
+  /// Wait before the next address starts while earlier attempts still run.
+  pub stagger: Duration,
+  /// Limit for one attempt while more addresses wait.
+  pub attempt_timeout: Duration,
+  /// The dial fails at this instant.
+  pub deadline: Instant,
+}
+
+/// Why no attempt connected.
+#[derive(Debug)]
+pub(crate) struct RaceFailure {
+  pub attempts: usize,
+  pub families: Families,
+  pub deadline_hit: bool,
+  pub last: Option<io::Error>,
+}
+
+impl RaceFailure {
+  /// One error for the log. `what` names the dial: `direct connect to h:443`.
+  /// It never names an address: a log line must not list visited sites.
+  pub(crate) fn into_error(self, what: &str, limit: Duration) -> io::Error {
+    let tail = format!("attempts={} families={}", self.attempts, self.families);
+    match self.last {
+      _ if self.deadline_hit => io::Error::new(
+        io::ErrorKind::TimedOut,
+        format!("{what} timed out after {limit:?} {tail}"),
+      ),
+      Some(last) => io::Error::new(last.kind(), format!("{what} failed: {last} {tail}")),
+      None => io::Error::new(
+        io::ErrorKind::InvalidInput,
+        format!("{what} failed: no address {tail}"),
+      ),
+    }
+  }
+}
+
+/// RFC 8305 order: the families alternate, starting with the family of the
+/// first address.
+fn interleave(addrs: Vec<SocketAddr>) -> Vec<SocketAddr> {
+  let Some(first) = addrs.first() else {
+    return addrs;
+  };
+  let lead_is_v6 = first.is_ipv6();
+  let (lead, other): (Vec<_>, Vec<_>) = addrs
+    .into_iter()
+    .partition(|addr| addr.is_ipv6() == lead_is_v6);
+  let mut order = Vec::with_capacity(lead.len() + other.len());
+  let mut lead = lead.into_iter();
+  let mut other = other.into_iter();
+  loop {
+    match (lead.next(), other.next()) {
+      (None, None) => return order,
+      (a, b) => order.extend(a.into_iter().chain(b)),
+    }
+  }
+}
+
+/// Run `connect` across `addrs` and return the first connection with its
+/// address. The losing attempts are dropped.
+pub(crate) async fn race<T, F, Fut>(
+  addrs: Vec<SocketAddr>,
+  plan: &Race,
+  mut connect: F,
+) -> Result<(T, SocketAddr), RaceFailure>
+where
+  F: FnMut(SocketAddr) -> Fut,
+  Fut: Future<Output = io::Result<T>>,
+{
+  let mut waiting: VecDeque<SocketAddr> = interleave(addrs).into();
+  let mut running = FuturesUnordered::new();
+  let mut failure = RaceFailure {
+    attempts: 0,
+    families: Families::default(),
+    deadline_hit: false,
+    last: None,
+  };
+  let mut next_start = Instant::now();
+  loop {
+    let now = Instant::now();
+    if running.len() < MAX_IN_FLIGHT && (running.is_empty() || now >= next_start) {
+      if let Some(addr) = waiting.pop_front() {
+        failure.attempts += 1;
+        failure.families.add(addr);
+        let limit = if waiting.is_empty() {
+          plan.deadline
+        } else {
+          (now + plan.attempt_timeout).min(plan.deadline)
+        };
+        let attempt = connect(addr);
+        running.push(async move { (addr, tokio::time::timeout_at(limit, attempt).await) });
+        next_start = now + plan.stagger;
+        continue;
+      }
+    }
+    if running.is_empty() {
+      failure.deadline_hit = Instant::now() >= plan.deadline;
+      return Err(failure);
+    }
+    let may_start = running.len() < MAX_IN_FLIGHT && !waiting.is_empty();
+    tokio::select! {
+      Some((addr, outcome)) = running.next() => {
+        let error = match outcome {
+          Ok(Ok(value)) => return Ok((value, addr)),
+          Ok(Err(e)) => e,
+          Err(_) => io::Error::new(io::ErrorKind::TimedOut, "attempt timed out"),
+        };
+        log::debug!("Dial attempt failed addr={addr} err=\"{error}\"");
+        failure.last = Some(error);
+        next_start = Instant::now();
+      }
+      () = tokio::time::sleep_until(next_start), if may_start => {}
+      () = tokio::time::sleep_until(plan.deadline) => {
+        failure.deadline_hit = true;
+        return Err(failure);
+      }
+    }
+  }
+}
+
+/// Race TCP connections to `addrs` with the timings of RFC 8305.
+pub(crate) async fn connect_addrs(
+  what: &str,
+  addrs: Vec<SocketAddr>,
+  deadline: Instant,
+  limit: Duration,
+) -> io::Result<(TcpStream, SocketAddr)> {
+  let plan = Race {
+    stagger: ATTEMPT_DELAY,
+    attempt_timeout: ATTEMPT_TIMEOUT,
+    deadline,
+  };
+  match race(addrs, &plan, TcpStream::connect).await {
+    Ok((stream, addr)) => {
+      let _ = stream.set_nodelay(true);
+      Ok((stream, addr))
+    }
+    Err(failure) => Err(failure.into_error(what, limit)),
+  }
+}
+
+/// Connect to `host:port` within `limit`, name lookup included. `role` starts
+/// the error text: `direct`, `upstream proxy`.
+pub(crate) async fn connect_host(
+  role: &str,
+  host: &str,
+  port: u16,
+  limit: Duration,
+) -> io::Result<TcpStream> {
+  connect_host_with(HostResolver::system(), role, host, port, limit).await
+}
+
+async fn connect_host_with(
+  resolver: &HostResolver,
+  role: &str,
+  host: &str,
+  port: u16,
+  limit: Duration,
+) -> io::Result<TcpStream> {
+  let deadline = Instant::now() + limit;
+  let host = host
+    .strip_prefix('[')
+    .and_then(|h| h.strip_suffix(']'))
+    .unwrap_or(host);
+  let what = format!(
+    "{role} connect to {}",
+    crate::proxy_server::connect_authority(host, port)
+  );
+  if let Ok(ip) = host.parse::<IpAddr>() {
+    let addrs = vec![SocketAddr::new(ip, port)];
+    return Ok(connect_addrs(&what, addrs, deadline, limit).await?.0);
+  }
+  let addrs = resolver
+    .resolve(host, limit)
+    .await?
+    .iter()
+    .map(|addr| {
+      let mut addr = *addr;
+      addr.set_port(port);
+      addr
+    })
+    .collect();
+  let (stream, winner) = connect_addrs(&what, addrs, deadline, limit).await?;
+  resolver.prefer(host, winner.ip());
+  Ok(stream)
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use std::sync::atomic::AtomicUsize;
+
+  fn plan(attempt_timeout: Duration, limit: Duration) -> Race {
+    Race {
+      stagger: ATTEMPT_DELAY,
+      attempt_timeout,
+      deadline: Instant::now() + limit,
+    }
+  }
+
+  /// A port on this machine where nothing listens.
+  async fn closed_port() -> SocketAddr {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    listener.local_addr().unwrap()
+  }
+
+  fn resolver_of(
+    addrs: Vec<SocketAddr>,
+    calls: Arc<AtomicUsize>,
+    fail_first: bool,
+  ) -> HostResolver {
+    HostResolver::new(Arc::new(
+      move |_host: String| -> BoxFuture<'static, io::Result<Vec<SocketAddr>>> {
+        let call = calls.fetch_add(1, Ordering::SeqCst);
+        let addrs = addrs.clone();
+        Box::pin(async move {
+          tokio::time::sleep(Duration::from_millis(100)).await;
+          if fail_first && call == 0 {
+            Err(io::Error::other("resolver not ready"))
+          } else {
+            Ok(addrs)
+          }
+        })
+      },
+    ))
+  }
+
+  #[test]
+  fn families_alternate_from_the_first_answer() {
+    let v6 = |n: u16| {
+      SocketAddr::new(
+        IpAddr::V6(std::net::Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, n)),
+        443,
+      )
+    };
+    let v4 = |n: u8| SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, n)), 443);
+    assert_eq!(
+      interleave(vec![v6(1), v6(2), v6(3), v4(1), v4(2)]),
+      vec![v6(1), v4(1), v6(2), v4(2), v6(3)]
+    );
+    assert_eq!(
+      interleave(vec![v4(1), v4(2), v6(1)]),
+      vec![v4(1), v6(1), v4(2)]
+    );
+    assert!(interleave(Vec::new()).is_empty());
+  }
+
+  #[tokio::test]
+  async fn an_address_that_never_answers_costs_one_delay() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let good = listener.local_addr().unwrap();
+    let black_hole: SocketAddr = "192.0.2.1:9".parse().unwrap();
+    let started = std::time::Instant::now();
+    let (_, winner) = race(
+      vec![black_hole, good],
+      &plan(Duration::from_secs(4), Duration::from_secs(10)),
+      |addr| async move {
+        if addr == black_hole {
+          std::future::pending::<()>().await;
+        }
+        TcpStream::connect(addr).await
+      },
+    )
+    .await
+    .expect("the second address connects");
+    assert_eq!(winner, good);
+    assert!(
+      started.elapsed() < Duration::from_secs(1),
+      "{:?}",
+      started.elapsed()
+    );
+  }
+
+  #[tokio::test]
+  async fn a_real_unreachable_first_address_does_not_use_the_whole_limit() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let good = listener.local_addr().unwrap();
+    let started = std::time::Instant::now();
+    let (_, winner) = connect_addrs(
+      "direct connect to test",
+      vec!["10.255.255.1:9".parse().unwrap(), good],
+      Instant::now() + Duration::from_secs(10),
+      Duration::from_secs(10),
+    )
+    .await
+    .expect("the local listener connects");
+    assert_eq!(winner, good);
+    assert!(
+      started.elapsed() < Duration::from_secs(2),
+      "{:?}",
+      started.elapsed()
+    );
+  }
+
+  #[tokio::test]
+  async fn when_every_address_fails_the_error_counts_the_attempts() {
+    let addrs = vec![closed_port().await, closed_port().await];
+    let error = connect_addrs(
+      "direct connect to test",
+      addrs,
+      Instant::now() + Duration::from_secs(5),
+      Duration::from_secs(5),
+    )
+    .await
+    .expect_err("nothing listens");
+    let text = error.to_string();
+    assert!(text.contains("attempts=2"), "{text}");
+    assert!(text.contains("families=ipv4"), "{text}");
+    assert_ne!(error.kind(), io::ErrorKind::TimedOut, "{text}");
+
+    let silent: Vec<SocketAddr> = vec![
+      "192.0.2.1:9".parse().unwrap(),
+      "[2001:db8::1]:9".parse().unwrap(),
+    ];
+    let failure = race(
+      silent,
+      &plan(Duration::from_millis(200), Duration::from_millis(600)),
+      |_| std::future::pending::<io::Result<()>>(),
+    )
+    .await
+    .expect_err("nothing answers");
+    assert_eq!(failure.attempts, 2);
+    assert!(failure.deadline_hit);
+    let error = failure.into_error("direct connect to test", Duration::from_millis(600));
+    assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+    assert!(error.to_string().contains("families=ipv4,ipv6"), "{error}");
+  }
+
+  #[tokio::test]
+  async fn concurrent_dials_of_one_host_share_one_lookup() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let calls = Arc::new(AtomicUsize::new(0));
+    let resolver = resolver_of(vec!["127.0.0.1:0".parse().unwrap()], calls.clone(), false);
+    let dials = (0..16).map(|_| {
+      let resolver = resolver.clone();
+      tokio::spawn(async move {
+        connect_host_with(
+          &resolver,
+          "direct",
+          "Example.Test",
+          port,
+          Duration::from_secs(5),
+        )
+        .await
+      })
+    });
+    for dial in futures_util::future::join_all(dials).await {
+      dial.unwrap().expect("dial through the shared lookup");
+    }
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    connect_host_with(
+      &resolver,
+      "direct",
+      "example.test",
+      port,
+      Duration::from_secs(5),
+    )
+    .await
+    .expect("cached answer");
+    assert_eq!(calls.load(Ordering::SeqCst), 1, "the answer is reused");
+  }
+
+  #[tokio::test]
+  async fn a_failed_dial_does_not_fail_the_next_dial_of_the_host() {
+    let calls = Arc::new(AtomicUsize::new(0));
+    let resolver = resolver_of(vec!["127.0.0.1:0".parse().unwrap()], calls.clone(), true);
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    connect_host_with(
+      &resolver,
+      "direct",
+      "example.test",
+      port,
+      Duration::from_secs(5),
+    )
+    .await
+    .expect_err("the first lookup fails");
+    connect_host_with(
+      &resolver,
+      "direct",
+      "example.test",
+      port,
+      Duration::from_secs(5),
+    )
+    .await
+    .expect("a failed lookup is not remembered");
+    assert_eq!(calls.load(Ordering::SeqCst), 2);
+
+    let refused = closed_port().await.port();
+    let error = connect_host_with(
+      &resolver,
+      "direct",
+      "example.test",
+      refused,
+      Duration::from_secs(5),
+    )
+    .await
+    .expect_err("nothing listens");
+    assert_eq!(error.kind(), io::ErrorKind::ConnectionRefused, "{error}");
+    let started = std::time::Instant::now();
+    connect_host_with(
+      &resolver,
+      "direct",
+      "example.test",
+      port,
+      Duration::from_secs(5),
+    )
+    .await
+    .expect("a failed connect is not remembered");
+    assert!(started.elapsed() < Duration::from_secs(1));
+  }
+
+  #[tokio::test]
+  async fn the_address_that_answered_goes_first_next_time() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let calls = Arc::new(AtomicUsize::new(0));
+    let resolver = resolver_of(
+      vec![
+        "127.0.0.2:0".parse().unwrap(),
+        "127.0.0.1:0".parse().unwrap(),
+      ],
+      calls,
+      false,
+    );
+    // 127.0.0.2 refuses on Linux and times out on macOS; either way the
+    // second address wins.
+    connect_host_with(
+      &resolver,
+      "direct",
+      "example.test",
+      port,
+      Duration::from_secs(5),
+    )
+    .await
+    .expect("the second address connects");
+    let order = resolver
+      .resolve("example.test", LOOKUP_TIMEOUT)
+      .await
+      .unwrap();
+    assert_eq!(order[0].ip(), "127.0.0.1".parse::<IpAddr>().unwrap());
+  }
+
+  #[tokio::test]
+  async fn ip_literals_skip_the_lookup() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let calls = Arc::new(AtomicUsize::new(0));
+    let resolver = resolver_of(Vec::new(), calls.clone(), false);
+    connect_host_with(
+      &resolver,
+      "direct",
+      "127.0.0.1",
+      port,
+      Duration::from_secs(5),
+    )
+    .await
+    .expect("literal");
+    assert_eq!(calls.load(Ordering::SeqCst), 0);
+  }
+}

src-tauri/src/lib.rs

@@ -133,6 +133,7 @@ mod browser_version_manager;
 mod cdp_target;
 mod checksum;
 mod default_browser;
+mod dialer;
 pub mod dns_blocklist;
 mod downloaded_browsers_registry;
 mod downloader;

src-tauri/src/proxy_server.rs

@@ -469,15 +469,20 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for SocksHandshakeLogger<S> {
 }
 
 async fn connect_via_socks(
-  socks_addr: &str,
+  socks_host: &str,
+  socks_port: u16,
   target_host: &str,
   target_port: u16,
   is_socks5: bool,
   auth: Option<(&str, &str)>,
 ) -> Result<TcpStream, Box<dyn std::error::Error>> {
-  let stream = tokio::time::timeout(UPSTREAM_DIAL_TIMEOUT, TcpStream::connect(socks_addr))
-    .await
-    .map_err(|_| format!("SOCKS upstream connect to {socks_addr} timed out"))??;
+  let stream = crate::dialer::connect_host(
+    "SOCKS upstream",
+    socks_host,
+    socks_port,
+    UPSTREAM_DIAL_TIMEOUT,
+  )
+  .await?;
 
   if is_socks5 {
     // SOCKS5 connection using async_socks5
@@ -496,8 +501,9 @@ async fn connect_via_socks(
 
     let has_auth = auth_info.is_some();
     log::trace!(
-      "[socks-handshake] dialing {} (target={}:{}, has_auth={})",
-      socks_addr,
+      "[socks-handshake] dialing {}:{} (target={}:{}, has_auth={})",
+      socks_host,
+      socks_port,
       target_host,
       target_port,
       has_auth
@@ -514,7 +520,7 @@ async fn connect_via_socks(
     // and silently FIN instead of returning an RFC1929 status. BufStream
     // coalesces those small writes into one syscall on flush — this is
     // the usage pattern shown in the async_socks5 README
-    let label = format!("{socks_addr}->{target_host}:{target_port}");
+    let label = format!("{socks_host}:{socks_port}->{target_host}:{target_port}");
     let logged = SocksHandshakeLogger::new(stream, label);
     let mut buffered = tokio::io::BufStream::new(logged);
     let handshake = tokio::time::timeout(
@@ -878,7 +884,6 @@ async fn handle_http_via_socks4(
 
   let socks_host = upstream.host_str().unwrap_or("127.0.0.1");
   let socks_port = upstream.port().unwrap_or(1080);
-  let socks_addr = format!("{}:{}", socks_host, socks_port);
 
   // Parse target from request URI
   let target_uri = req.uri();
@@ -886,26 +891,28 @@ async fn handle_http_via_socks4(
   let target_port = target_uri.port_u16().unwrap_or(80);
 
   // Connect to SOCKS4 proxy
-  let mut socks_stream =
-    match tokio::time::timeout(UPSTREAM_DIAL_TIMEOUT, TcpStream::connect(&socks_addr)).await {
-      Ok(Ok(stream)) => stream,
-      Ok(Err(e)) => {
-        warn_upstream_failure("SOCKS4 upstream connect failed", &e.to_string(), None);
-        let mut response = Response::new(Full::new(Bytes::from(format!(
-          "Failed to connect to SOCKS4 proxy: {}",
-          e
-        ))));
-        *response.status_mut() = StatusCode::BAD_GATEWAY;
-        return Ok(response);
-      }
-      Err(_) => {
-        warn_upstream_failure("SOCKS4 upstream connect failed", "timed out", None);
-        let mut response =
-          Response::new(Full::new(Bytes::from("Connect to SOCKS4 proxy timed out")));
-        *response.status_mut() = StatusCode::GATEWAY_TIMEOUT;
-        return Ok(response);
-      }
-    };
+  let mut socks_stream = match crate::dialer::connect_host(
+    "SOCKS4 upstream",
+    socks_host,
+    socks_port,
+    UPSTREAM_DIAL_TIMEOUT,
+  )
+  .await
+  {
+    Ok(stream) => stream,
+    Err(e) => {
+      warn_upstream_failure("SOCKS4 upstream connect failed", &e.to_string(), None);
+      let mut response = Response::new(Full::new(Bytes::from(format!(
+        "Failed to connect to SOCKS4 proxy: {e}"
+      ))));
+      *response.status_mut() = if e.kind() == io::ErrorKind::TimedOut {
+        StatusCode::GATEWAY_TIMEOUT
+      } else {
+        StatusCode::BAD_GATEWAY
+      };
+      return Ok(response);
+    }
+  };
 
   // Build a SOCKS4a CONNECT request. We deliberately do NOT resolve the target
   // hostname locally: tokio::net::lookup_host would call the HOST resolver
@@ -1172,6 +1179,29 @@ async fn forward_http_over_stream<S: AsyncRead + AsyncWrite + Unpin>(
   Ok(hyper_response)
 }
 
+/// Dial the Shadowsocks server and open a tunnel to `target` over it. The
+/// tunnel header goes out with the first write.
+async fn connect_via_shadowsocks(
+  ss_host: &str,
+  ss_port: u16,
+  svr_cfg: &shadowsocks::config::ServerConfig,
+  target: shadowsocks::relay::Address,
+) -> io::Result<shadowsocks::relay::tcprelay::proxy_stream::ProxyClientStream<TcpStream>> {
+  let tcp = crate::dialer::connect_host(
+    "Shadowsocks server",
+    ss_host,
+    ss_port,
+    UPSTREAM_DIAL_TIMEOUT,
+  )
+  .await?;
+  let context = shadowsocks::context::Context::new_shared(shadowsocks::config::ServerType::Local);
+  Ok(
+    shadowsocks::relay::tcprelay::proxy_stream::ProxyClientStream::from_stream(
+      context, tcp, svr_cfg, target,
+    ),
+  )
+}
+
 /// Handle plain HTTP requests through a Shadowsocks upstream.
 /// reqwest doesn't support SS natively, so we connect through the SS tunnel
 /// manually and forward the HTTP request/response.
@@ -1206,7 +1236,6 @@ async fn handle_http_via_shadowsocks(
     }
   };
 
-  let context = shadowsocks::context::Context::new_shared(shadowsocks::config::ServerType::Local);
   let svr_cfg = match shadowsocks::config::ServerConfig::new(
     shadowsocks::config::ServerAddr::from((ss_host.to_string(), ss_port)),
     &password,
@@ -1222,13 +1251,7 @@ async fn handle_http_via_shadowsocks(
 
   let target_addr = shadowsocks::relay::Address::DomainNameAddress(domain.clone(), port);
 
-  let mut stream = match shadowsocks::relay::tcprelay::proxy_stream::ProxyClientStream::connect(
-    context,
-    &svr_cfg,
-    target_addr,
-  )
-  .await
-  {
+  let mut stream = match connect_via_shadowsocks(ss_host, ss_port, &svr_cfg, target_addr).await {
     Ok(s) => s,
     Err(e) => {
       let mut resp = Response::new(Full::new(Bytes::from(format!("SS connect: {e}"))));
@@ -2097,14 +2120,7 @@ async fn handle_connect_from_buffer(
     return Err("Invalid CONNECT request".into());
   }
 
-  let target = parts[1];
-  let (target_host, target_port) = if let Some(colon_pos) = target.find(':') {
-    let host = &target[..colon_pos];
-    let port: u16 = target[colon_pos + 1..].parse().unwrap_or(443);
-    (host, port)
-  } else {
-    (target, 443)
-  };
+  let (target_host, target_port) = split_authority(parts[1]);
 
   // Block if domain is in the DNS blocklist (before any connection)
   if blocklist_matcher.is_blocked(target_host) {
@@ -2217,11 +2233,17 @@ fn worker_connection_limit() -> usize {
   }
 }
 
-/// Connect timeout for the direct (no-upstream) dial path. Bounds a wedged
-/// `getaddrinfo` so a broken resolver can't park a blocking thread for the
-/// full OS timeout.
+/// Limit for a direct (no-upstream) dial, name lookup included.
 const DIRECT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 
+async fn dial_direct(host: &str, port: u16) -> io::Result<TcpStream> {
+  crate::dialer::connect_host("direct", host, port, DIRECT_CONNECT_TIMEOUT).await
+}
+
+/// With a profile DNS server and an upstream, the wait before the second
+/// address of a name gets its own dial through the upstream.
+const UPSTREAM_ATTEMPT_DELAY: std::time::Duration = std::time::Duration::from_secs(2);
+
 /// Overall timeout for dialing an UPSTREAM proxy (TCP connect + CONNECT/SOCKS/SS
 /// handshake). Without it, an upstream that accepts TCP but stalls before
 /// replying hangs the worker task forever and holds a connection slot; under
@@ -2249,91 +2271,6 @@ const MAX_HTTP_RESPONSE_BUFFER: usize = 10 * 1024 * 1024;
 /// the body and break all three.
 const PLAIN_HTTP_EXCHANGE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
 
-/// Per-host failure state (last failure instant, consecutive failure count) for
-/// the direct dial path. Process-global — each worker is its own process.
-fn direct_dial_failures() -> &'static Mutex<HashMap<String, (std::time::Instant, u32)>> {
-  static M: OnceLock<Mutex<HashMap<String, (std::time::Instant, u32)>>> = OnceLock::new();
-  M.get_or_init(|| Mutex::new(HashMap::new()))
-}
-
-/// If `host` is inside its failure backoff window, return the remaining time so
-/// the caller can short-circuit without a fresh getaddrinfo/connect. Never
-/// mutates state, so the window always expires and the path self-heals once
-/// DNS recovers.
-fn direct_backoff_remaining(host: &str) -> Option<std::time::Duration> {
-  let map = direct_dial_failures();
-  let guard = map.lock().unwrap();
-  let (last, fails) = guard.get(host).copied()?;
-  // Exponential window capped at 30s: 2, 4, 8, 16, 30, 30, ...
-  let window = std::time::Duration::from_secs((1u64 << fails.min(5)).min(30));
-  let elapsed = last.elapsed();
-  if elapsed < window {
-    Some(window - elapsed)
-  } else {
-    None
-  }
-}
-
-/// Record a direct-dial failure for `host`, growing its backoff window.
-fn direct_backoff_record(host: &str) {
-  let map = direct_dial_failures();
-  let mut guard = map.lock().unwrap();
-  // Bound memory against a page that emits many distinct failing hosts.
-  if guard.len() > 2048 {
-    guard.retain(|_, (last, _)| last.elapsed() < std::time::Duration::from_secs(60));
-  }
-  let entry = guard
-    .entry(host.to_string())
-    .or_insert_with(|| (std::time::Instant::now(), 0));
-  entry.0 = std::time::Instant::now();
-  entry.1 = entry.1.saturating_add(1);
-}
-
-/// Clear `host`'s failure state after a successful dial.
-fn direct_backoff_clear(host: &str) {
-  direct_dial_failures().lock().unwrap().remove(host);
-}
-
-/// Dial a target directly (no upstream) with a connect timeout and per-host
-/// failure backoff. This is the server-side counterpart to the browser's
-/// instant client-side retry: when a host's DNS/connect is failing (e.g. the
-/// macOS resolver wedges after sleep/wake), repeated CONNECT requests
-/// short-circuit
-/// here instead of each spawning a fresh blocking getaddrinfo — which is what
-/// let a retry-storm exhaust the blocking thread pool and peg the CPU.
-async fn dial_direct(host: &str, port: u16) -> Result<TcpStream, Box<dyn std::error::Error>> {
-  if let Some(remaining) = direct_backoff_remaining(host) {
-    return Err(
-      format!(
-        "skipping direct dial to {host}: backing off ~{}s after repeated connect failures",
-        remaining.as_secs().max(1)
-      )
-      .into(),
-    );
-  }
-  match tokio::time::timeout(DIRECT_CONNECT_TIMEOUT, TcpStream::connect((host, port))).await {
-    Ok(Ok(stream)) => {
-      let _ = stream.set_nodelay(true);
-      direct_backoff_clear(host);
-      Ok(stream)
-    }
-    Ok(Err(e)) => {
-      direct_backoff_record(host);
-      Err(e.into())
-    }
-    Err(_) => {
-      direct_backoff_record(host);
-      Err(
-        format!(
-          "direct connect to {host}:{port} timed out after {}s",
-          DIRECT_CONNECT_TIMEOUT.as_secs()
-        )
-        .into(),
-      )
-    }
-  }
-}
-
 /// Host of a buffered `CONNECT host:port HTTP/1.1` request.
 fn connect_target_host(request: &[u8]) -> Option<String> {
   let line_end = request
@@ -2341,14 +2278,23 @@ fn connect_target_host(request: &[u8]) -> Option<String> {
     .position(|&b| b == b'\r' || b == b'\n')
     .unwrap_or(request.len());
   let line = String::from_utf8_lossy(&request[..line_end]);
-  let target = line.split_whitespace().nth(1)?;
-  let host = match target.strip_prefix('[') {
-    Some(rest) => rest.split(']').next()?,
-    None => target.rsplit_once(':').map_or(target, |(host, _)| host),
-  };
+  let (host, _) = split_authority(line.split_whitespace().nth(1)?);
   (!host.is_empty()).then(|| host.to_string())
 }
 
+/// Host and port of a CONNECT authority, `host:port` or `[v6]:port`. The port
+/// defaults to 443.
+fn split_authority(target: &str) -> (&str, u16) {
+  if let Some((host, rest)) = target.strip_prefix('[').and_then(|t| t.split_once(']')) {
+    let port = rest.strip_prefix(':').and_then(|p| p.parse().ok());
+    return (host, port.unwrap_or(443));
+  }
+  match target.split_once(':') {
+    Some((host, port)) => (host, port.parse().unwrap_or(443)),
+    None => (target, 443),
+  }
+}
+
 /// Warn about a failed tunnel or proxied request, at most once per distinct
 /// message every 30s. The target host is cut out of the message: these lines
 /// reach the file, and the file must not list the sites a profile visited.
@@ -2414,6 +2360,17 @@ async fn read_upstream_connect_response<S: AsyncRead + Unpin>(
   }
 }
 
+/// `host:port` for a CONNECT line. An IPv6 literal (a SOCKS5 client sends one
+/// for `http://[2001:db8::1]/`) needs brackets, or the proxy reads its last
+/// group as the port.
+pub(crate) fn connect_authority(host: &str, port: u16) -> String {
+  if host.parse::<std::net::Ipv6Addr>().is_ok() {
+    format!("[{host}]:{port}")
+  } else {
+    format!("{host}:{port}")
+  }
+}
+
 /// Perform the HTTP CONNECT handshake over an already-established hop to the
 /// proxy and return the tunnelled stream.
 ///
@@ -2422,17 +2379,6 @@ async fn read_upstream_connect_response<S: AsyncRead + Unpin>(
 /// only place `Proxy-Authorization` is written, so whether those credentials
 /// cross the network in the clear is decided entirely by which stream the
 /// caller hands in, nothing here can weaken it.
-/// `host:port` for a CONNECT line. An IPv6 literal (a SOCKS5 client sends one
-/// for `http://[2001:db8::1]/`) needs brackets, or the proxy reads its last
-/// group as the port.
-fn connect_authority(host: &str, port: u16) -> String {
-  if host.parse::<std::net::Ipv6Addr>().is_ok() {
-    format!("[{host}]:{port}")
-  } else {
-    format!("{host}:{port}")
-  }
-}
-
 async fn connect_via_http_proxy<S: AsyncStream + 'static>(
   mut proxy_stream: S,
   proxy_host: &str,
@@ -2544,16 +2490,50 @@ pub(crate) async fn connect_to_target_via_upstream(
   let Some(resolver) = resolver.filter(|_| !should_bypass) else {
     return dial_route(target_host, target_port, upstream_url, should_bypass).await;
   };
-  let addresses = resolver.resolve(target_host).await?;
-  // A second address covers an exit that cannot reach the first one.
-  let mut last_error = String::new();
-  for address in addresses.iter().take(2) {
-    match dial_route(&address.to_string(), target_port, upstream_url, false).await {
-      Ok(stream) => return Ok(stream),
-      Err(e) => last_error = e.to_string(),
-    }
+  let addresses: Vec<SocketAddr> = resolver
+    .resolve(target_host)
+    .await?
+    .iter()
+    .map(|ip| SocketAddr::new(*ip, target_port))
+    .collect();
+  let authority = connect_authority(target_host, target_port);
+  if matches!(upstream_url, None | Some("DIRECT")) {
+    let (stream, _) = crate::dialer::connect_addrs(
+      &format!("direct connect to {authority}"),
+      addresses,
+      tokio::time::Instant::now() + DIRECT_CONNECT_TIMEOUT,
+      DIRECT_CONNECT_TIMEOUT,
+    )
+    .await?;
+    return Ok(Box::new(stream));
+  }
+  // Each attempt is a whole dial through the upstream, so the next address
+  // waits longer, and only a second address gets a turn: it covers an exit
+  // that cannot reach the first one.
+  let limit = 2 * UPSTREAM_DIAL_TIMEOUT;
+  let plan = crate::dialer::Race {
+    stagger: UPSTREAM_ATTEMPT_DELAY,
+    attempt_timeout: UPSTREAM_DIAL_TIMEOUT,
+    deadline: tokio::time::Instant::now() + limit,
+  };
+  let candidates = addresses.into_iter().take(2).collect();
+  let raced = crate::dialer::race(candidates, &plan, |address| async move {
+    dial_route(&address.ip().to_string(), target_port, upstream_url, false)
+      .await
+      .map_err(|e| io::Error::other(e.to_string()))
+  })
+  .await;
+  match raced {
+    Ok((stream, _)) => Ok(stream),
+    Err(failure) => Err(
+      failure
+        .into_error(
+          &format!("connect to {authority} through the upstream"),
+          limit,
+        )
+        .into(),
+    ),
   }
-  Err(last_error.into())
 }
 
 /// Dial `target_host:target_port` over the route as given, with no name
@@ -2564,10 +2544,6 @@ pub(crate) async fn dial_route(
   upstream_url: Option<&str>,
   should_bypass: bool,
 ) -> Result<BoxedAsyncStream, Box<dyn std::error::Error>> {
-  // Helper: configure outbound TCP to match browser TCP fingerprint
-  let configure_tcp = |stream: &TcpStream| {
-    let _ = stream.set_nodelay(true);
-  };
   let target_stream: BoxedAsyncStream = match upstream_url {
     None | Some("DIRECT") => Box::new(dial_direct(target_host, target_port).await?),
     _ if should_bypass => Box::new(dial_direct(target_host, target_port).await?),
@@ -2585,15 +2561,13 @@ pub(crate) async fn dial_route(
           "http" | "https" => {
             let proxy_host = upstream.host_str().unwrap_or("127.0.0.1");
             let proxy_port = upstream.port().unwrap_or(8080);
-            let proxy_stream = tokio::time::timeout(
+            let proxy_stream = crate::dialer::connect_host(
+              "upstream proxy",
+              proxy_host,
+              proxy_port,
               UPSTREAM_DIAL_TIMEOUT,
-              TcpStream::connect((proxy_host, proxy_port)),
             )
-            .await
-            .map_err(|_| {
-              format!("upstream proxy connect to {proxy_host}:{proxy_port} timed out")
-            })??;
-            configure_tcp(&proxy_stream);
+            .await?;
 
             connect_via_http_proxy(
               proxy_stream,
@@ -2611,15 +2585,13 @@ pub(crate) async fn dial_route(
           "httpstls" => {
             let proxy_host = upstream.host_str().unwrap_or("127.0.0.1");
             let proxy_port = upstream.port().unwrap_or(443);
-            let tcp = tokio::time::timeout(
+            let tcp = crate::dialer::connect_host(
+              "upstream proxy",
+              proxy_host,
+              proxy_port,
               UPSTREAM_DIAL_TIMEOUT,
-              TcpStream::connect((proxy_host, proxy_port)),
             )
-            .await
-            .map_err(|_| {
-              format!("upstream proxy connect to {proxy_host}:{proxy_port} timed out")
-            })??;
-            configure_tcp(&tcp);
+            .await?;
 
             let tls = tls_wrap_upstream_hop(tcp, proxy_host).await?;
 
@@ -2636,13 +2608,13 @@ pub(crate) async fn dial_route(
           "socks4" | "socks5" => {
             let socks_host = upstream.host_str().unwrap_or("127.0.0.1");
             let socks_port = upstream.port().unwrap_or(1080);
-            let socks_addr = format!("{}:{}", socks_host, socks_port);
 
             let (username, password) = upstream_userpass(&upstream);
             let auth = (!username.is_empty()).then_some((username.as_str(), password.as_str()));
 
             let stream = connect_via_socks(
-              &socks_addr,
+              socks_host,
+              socks_port,
               target_host,
               target_port,
               scheme == "socks5",
@@ -2677,8 +2649,6 @@ pub(crate) async fn dial_route(
               format!("Unsupported Shadowsocks cipher: {method_str}. Use e.g. aes-256-gcm, chacha20-ietf-poly1305, aes-128-gcm")
             })?;
 
-            let context =
-              shadowsocks::context::Context::new_shared(shadowsocks::config::ServerType::Local);
             let svr_cfg = shadowsocks::config::ServerConfig::new(
               shadowsocks::config::ServerAddr::from((ss_host.to_string(), ss_port)),
               &password,
@@ -2693,19 +2663,7 @@ pub(crate) async fn dial_route(
               }
             };
 
-            let stream = tokio::time::timeout(
-              UPSTREAM_DIAL_TIMEOUT,
-              shadowsocks::relay::tcprelay::proxy_stream::ProxyClientStream::connect(
-                context,
-                &svr_cfg,
-                target_addr,
-              ),
-            )
-            .await
-            .map_err(|_| "Shadowsocks connection timed out".to_string())?
-            .map_err(|e| format!("Shadowsocks connection failed: {e}"))?;
-
-            Box::new(stream)
+            Box::new(connect_via_shadowsocks(ss_host, ss_port, &svr_cfg, target_addr).await?)
           }
           _ => {
             return Err(format!("Unsupported upstream proxy scheme: {}", scheme).into());
@@ -3026,7 +2984,7 @@ mod tests {
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap().to_string();
+    let port = listener.local_addr().unwrap().port();
     let server = tokio::spawn(async move {
       let (mut socket, _) = listener.accept().await.unwrap();
       let mut request = vec![0_u8; 8 + 1 + "example.com".len() + 1];
@@ -3038,7 +2996,7 @@ mod tests {
       request
     });
 
-    connect_via_socks(&address, "example.com", 443, false, None)
+    connect_via_socks("127.0.0.1", port, "example.com", 443, false, None)
       .await
       .expect("SOCKS4a connect");
     let request = server.await.unwrap();
@@ -3053,6 +3011,18 @@ mod tests {
     assert_eq!(connect_authority("example.com", 443), "example.com:443");
   }
 
+  #[test]
+  fn a_connect_target_with_an_ipv6_literal_keeps_its_address() {
+    assert_eq!(split_authority("[2001:db8::1]:8443"), ("2001:db8::1", 8443));
+    assert_eq!(split_authority("[2001:db8::1]"), ("2001:db8::1", 443));
+    assert_eq!(split_authority("example.com:80"), ("example.com", 80));
+    assert_eq!(split_authority("example.com"), ("example.com", 443));
+    assert_eq!(
+      connect_target_host(b"CONNECT [2001:db8::1]:443 HTTP/1.1\r\n\r\n").as_deref(),
+      Some("2001:db8::1")
+    );
+  }
+
   #[test]
   fn a_fully_qualified_name_with_a_trailing_dot_is_still_blocked() {
     let matcher = matcher_of(&["tracker.com"]);

src-tauri/src/socks5_local.rs

@@ -573,12 +573,13 @@ async fn associate_upstream(
     None
   };
 
-  let proxy_stream = tokio::time::timeout(
+  let proxy_stream = crate::dialer::connect_host(
+    "upstream SOCKS5",
+    host,
+    port,
     crate::proxy_server::UPSTREAM_DIAL_TIMEOUT,
-    TcpStream::connect((host, port)),
   )
-  .await
-  .map_err(|_| format!("upstream SOCKS5 connect to {host}:{port} timed out"))??;
+  .await?;
   let bind_sock = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await?;
   // association_addr None => 0.0.0.0:0 (we accept replies from any peer).
   let datagram = tokio::time::timeout(
@@ -792,7 +793,8 @@ async fn run_udp_relay_socks5(
 }
 
 /// Resolve a UDP destination to a concrete socket address, with the profile
-/// DNS server when there is one and the system resolver otherwise.
+/// DNS server when there is one and the system resolver otherwise. The direct
+/// egress socket is IPv4, so an IPv4 answer goes first.
 async fn resolve_addr(addr: &AddrKind, resolver: Option<&ProfileResolver>) -> Option<SocketAddr> {
   match (addr, resolver) {
     (AddrKind::Ip(s), _) => Some(*s),
@@ -802,10 +804,18 @@ async fn resolve_addr(addr: &AddrKind, resolver: Option<&ProfileResolver>) -> Op
       .ok()?
       .first()
       .map(|ip| SocketAddr::new(*ip, *port)),
-    (AddrKind::Domain(domain, port), None) => tokio::net::lookup_host(format!("{domain}:{port}"))
-      .await
-      .ok()
-      .and_then(|mut it| it.next()),
+    (AddrKind::Domain(domain, port), None) => {
+      let found = crate::dialer::HostResolver::system()
+        .resolve(domain, crate::dialer::LOOKUP_TIMEOUT)
+        .await
+        .ok()?;
+      let ip = found
+        .iter()
+        .find(|a| a.is_ipv4())
+        .or_else(|| found.first())?
+        .ip();
+      Some(SocketAddr::new(ip, *port))
+    }
   }
 }