From a0425324c04a3e5fed34a955fd2ea32ec83c3f70 Mon Sep 17 00:00:00 2001 From: chuan Date: Wed, 24 Jun 2026 21:50:19 +0800 Subject: [PATCH] feat: provide a easy way to search the best ip --- README.md | 22 +++ src/cli.rs | 30 ++++ src/commands.rs | 4 +- src/easy.rs | 467 ++++++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 2 + src/ping.rs | 5 +- src/report.rs | 25 ++- src/speed.rs | 62 ++++++- 8 files changed, 607 insertions(+), 10 deletions(-) create mode 100644 src/easy.rs diff --git a/README.md b/README.md index dcb741d..d8a69c9 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,12 @@ fast-xray --node-file node.txt -n 50 --lat-top 30 --speed-top 10 节点 URL 含 `&`,可直接加引号传入,或用 `--node-file` 从文件读取。结果写入 `result/result.txt`。 +只要一个够快的 IP,用 `easy`——边发现边测速,第一个达标即停: + +```bash +fast-xray easy --node-file node.txt --speed 5 +``` + 各阶段也可单独运行用于调试或定制。 ## 流水线 @@ -104,6 +110,22 @@ flowchart LR auto 使用各阶段默认的超时、并发、丢包等参数;需精调时单独运行子命令。 +### easy — `fast-xray easy [NODE] --speed [OPTIONS]` + +只要一个够快的 IP 就停。后台持续用 `ping + latency` 往池子(桶)里补充有效 IP;前台每轮并发筛一批(5 个),并发下就达标的直接命中,有潜力的(≥ `目标/5`)再单独确认,第一个达到 `--speed` 的即为结果,直接输出可复制的节点。 + +先直连国内镜像(清华 TUNA)测一次本机不走代理的下载速度,作为带宽上限:`--speed` 高于它直接判定不可能;镜像偶发不可用则跳过、照常搜索。 + +| 参数 | 默认 | 说明 | +| --- | --- | --- | +| `[NODE]` / `--node-file` | — | 节点 URL,或从文件读取 | +| `--speed` | 直连×80% | 目标下载速度 MB/s,返回首个达标 IP;不填则取直连测速的 80% | +| `--max` | 100 | 最多遍历多少个**有效** IP(通过 ping + latency);用尽仍无达标即「找不到」 | +| `-6, --ipv6` | off | 同时搜索 IPv6 | +| `-o, --output` | result | 输出目录(写 result.txt) | + +固定参数:ping 并发 100、≤200ms;latency 并发 10、≤200ms;测速并发筛 5 个再单独确认、每次上限 5s。 + ### ping — `fast-xray ping [OPTIONS]` | 参数 | 默认 | 说明 | diff --git a/src/cli.rs b/src/cli.rs index 835572c..38317fa 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -31,6 +31,8 @@ pub(crate) enum Command { Speed(SpeedArgs), /// Build importable vless:// nodes from a valid-IP list and a node. Export(ExportArgs), + /// Find one IP fast enough to hit a target speed, then stop. + Easy(EasyArgs), } #[derive(Args)] @@ -216,6 +218,34 @@ pub(crate) struct SpeedArgs { pub(crate) verbose: bool, } +#[derive(Args)] +pub(crate) struct EasyArgs { + /// Input vless:// node URL (or use --node-file). + pub(crate) node: Option, + + /// Read the vless:// node URL from a file (avoids shell escaping of `&`). + #[arg(long)] + pub(crate) node_file: Option, + + /// Target download speed in MB/s; returns the first IP that reaches it. + /// Omit to default to 80% of the measured direct (no-proxy) speed. + #[arg(long)] + pub(crate) speed: Option, + + /// Give up after this many valid IPs (those that pass both ping and + /// latency). Reaching it without a fast-enough IP means "not found". + #[arg(long, default_value_t = 100)] + pub(crate) max: usize, + + /// Also search IPv6 ranges (off by default). + #[arg(short = '6', long = "ipv6")] + pub(crate) ipv6: bool, + + /// Output directory. Writes /result.txt with the chosen node. + #[arg(short = 'o', long, default_value = "result")] + pub(crate) output: PathBuf, +} + #[derive(Args)] pub(crate) struct ExportArgs { /// Input vless:// node URL (or use --node-file). diff --git a/src/commands.rs b/src/commands.rs index e08c900..3044a08 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -11,8 +11,8 @@ use crate::cli::{AutoArgs, ExportArgs, LatencyArgs, PingArgs, SpeedArgs}; use crate::cloudflare::{self, Family}; use crate::ping::{self, Progress, ProbeStatus}; use crate::report::{ - LiveProgress, print_speed_table, print_table, read_export_rows, read_ip_list, read_latency_csv, - resolve_node, write_csv, write_file, write_ips, write_speed_csv, + LiveProgress, print_speed_table, print_table, read_export_rows, read_ip_list, + read_latency_csv, resolve_node, write_csv, write_file, write_ips, write_speed_csv, }; use crate::latency::{self, LatStatus}; use crate::speed; diff --git a/src/easy.rs b/src/easy.rs new file mode 100644 index 0000000..2bade99 --- /dev/null +++ b/src/easy.rs @@ -0,0 +1,467 @@ +//! `easy`: find one IP fast enough to hit a target speed, then stop. +//! +//! A producer keeps a bounded *bucket* of validated IPs (those that pass ping + +//! latency) topped up. A consumer pulls a batch and screens it concurrently: +//! under n-way contention each stream gets roughly link/n, so anything reaching +//! `target/n` is worth a solo confirm and anything already at `target` is a +//! guaranteed pass. The promising ones are then re-tested single-threaded (for +//! an accurate number) until one clears `target`. A live "bucket / recent" view +//! shows what's queued, what's being tested, and the last few finished results. + +use std::collections::VecDeque; +use std::net::IpAddr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use anyhow::{Result, anyhow}; +use console::style; +use futures::future::join_all; +use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; + +use crate::cli::EasyArgs; +use crate::cloudflare::{self, CfRanges, Family}; +use crate::latency; +use crate::ping; +use crate::report::{print_easy_found, resolve_node, write_file}; +use crate::speed; +use crate::vless::VlessNode; + +// Fixed knobs — the point of `easy` is to not expose these. +const POOL: usize = 10; // bucket capacity (testing + waiting), also the live row count +const BATCH: usize = 5; // IPs screened concurrently per round +const RECENT: usize = 5; // finished results kept on screen for review +const PING_CONCURRENCY: usize = 100; +const PING_TIMEOUT: f64 = 3.0; +const PING_MIN_MS: f64 = 10.0; +const PING_MAX_MS: f64 = 200.0; +const LAT_CONCURRENCY: usize = 10; +const LAT_TIMEOUT: f64 = 5.0; +const LAT_MAX_MS: f64 = 200.0; +const SPEED_TIMEOUT: f64 = 5.0; +const SPEED_BYTES: u64 = 50_000_000; // file is large; time/steady-state caps first +const MAX_BARREN_ROUNDS: usize = 8; // give up after this many empty discovery rounds +const DEFAULT_SPEED_FRACTION: f64 = 0.80; // default target = this × direct speed +const POLL: Duration = Duration::from_millis(50); // producer/consumer idle poll +const FRAME: Duration = Duration::from_millis(120); // render tick +const FRAMES: [&str; 10] = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]; + +/// One finished speed test, kept for the "recent" panel. +struct Finished { + ip: IpAddr, + mbs: Option, + pass: bool, +} + +/// Shared, observable bucket. A plain mutex (never held across `.await`) is +/// enough: producer and consumer poll it, the renderer reads snapshots. +struct Bucket { + waiting: VecDeque<(IpAddr, Duration)>, // validated, not yet speed-tested + testing: Vec, // the batch being screened / confirmed + recent: VecDeque, // last RECENT finished, oldest first + valid: usize, // total minted (gates --max + header) + producer_done: bool, + found: bool, +} + +type Shared = Arc>; + +pub(crate) async fn run(args: EasyArgs) -> Result<()> { + let node = Arc::new(resolve_node(&args.node_file, &args.node)?); + if matches!(args.speed, Some(s) if s <= 0.0) { + return Err(anyhow!("--speed must be greater than 0")); + } + + // 0. Direct (no-proxy) speed via a domestic mirror — the link ceiling, and + // the source of the default target when --speed is omitted. + let spinner = spin("Measuring direct (no-proxy) download speed…"); + let baseline = speed::measure_direct(secs(SPEED_TIMEOUT), SPEED_BYTES).await; + match &baseline { + Ok(mbs) => spinner.finish_with_message(format!( + "{} direct download speed: {} MB/s", + style("✓").green().bold(), + style(format!("{mbs:.2}")).cyan() + )), + Err(_) => spinner.finish_with_message(format!( + "{} direct speed unavailable (direct access blocked?)", + style("!").yellow().bold() + )), + } + + // Resolve the target: explicit --speed, else a fraction of the measured + // direct speed. Reject a target the local link can't reach; with neither a + // target nor a measured link there's nothing to aim for. + let (target, note) = match (args.speed, baseline) { + (Some(s), Ok(base)) if s > base => { + return Err(anyhow!( + "direct speed is only {base:.2} MB/s — can't find a node ≥ {s:.2} MB/s" + )); + } + (Some(s), _) => (s, String::new()), + (None, Ok(base)) => ( + base * DEFAULT_SPEED_FRACTION, + format!(" ({:.0}% of direct {base:.2})", DEFAULT_SPEED_FRACTION * 100.0), + ), + (None, Err(_)) => { + return Err(anyhow!( + "couldn't measure direct speed to pick a default target — pass --speed " + )); + } + }; + + let family = if args.ipv6 { Family::Both } else { Family::V4 }; + let spinner = spin("Fetching Cloudflare ranges…"); + let ranges = Arc::new(cloudflare::fetch_ranges(family).await?); + spinner.finish_with_message(format!( + "{} Cloudflare ranges: {} v4, {} v6", + style("✓").green().bold(), + ranges.v4.len(), + ranges.v6.len() + )); + eprintln!( + "{} target ≥ {} MB/s{}, up to {} valid IPs", + style("easy").bold().cyan(), + style(format!("{target:.2}")).cyan(), + style(note).dim(), + args.max + ); + + let shared: Shared = Arc::new(Mutex::new(Bucket { + waiting: VecDeque::new(), + testing: Vec::new(), + recent: VecDeque::new(), + valid: 0, + producer_done: false, + found: false, + })); + let producer = tokio::spawn(produce(ranges.clone(), node.clone(), args.max, args.ipv6, shared.clone())); + let stop = Arc::new(AtomicBool::new(false)); + let ui = tokio::spawn(render_loop(shared.clone(), BucketView::new(), target, args.max, stop.clone())); + + // Consumer: pull a batch, screen it concurrently, then confirm the + // promising ones single-threaded (for an accurate number). + let mut hit: Option<(IpAddr, Duration, f64)> = None; + let mut tested = 0usize; + // What the locked acquire step decided — kept tiny so the guard is released + // before any await. + enum Step { + Stop, + Wait, + Go(Vec<(IpAddr, Duration)>), + } + 'consume: loop { + let step = { + let mut b = shared.lock().unwrap(); + let n = b.waiting.len().min(BATCH); + if b.found || (n == 0 && b.producer_done) { + Step::Stop + } else if n == 0 { + Step::Wait + } else { + let batch: Vec<_> = b.waiting.drain(..n).collect(); + b.testing = batch.iter().map(|(ip, _)| *ip).collect(); + Step::Go(batch) + } + }; + let batch = match step { + Step::Stop => break, + Step::Wait => { + tokio::time::sleep(POLL).await; + continue; + } + Step::Go(batch) => batch, + }; + let n = batch.len(); + tested += n; + + // Screen the whole batch at once. + let screened = join_all(batch.into_iter().map(|(ip, lat)| { + let node = node.clone(); + async move { + let r = + speed::measure_download(node.as_ref(), ip, secs(SPEED_TIMEOUT), SPEED_BYTES).await; + (ip, lat, r) + } + })) + .await; + + // Anything at target/n is promising; below that (or failed) is dropped. + let floor = target / n as f64; + let mut candidates: Vec<(IpAddr, Duration, f64)> = Vec::new(); + { + let mut b = shared.lock().unwrap(); + for (ip, lat, r) in screened { + match r { + Ok(s) if s >= floor => candidates.push((ip, lat, s)), + Ok(s) => b.recent.push_back(Finished { ip, mbs: Some(s), pass: false }), + Err(_) => b.recent.push_back(Finished { ip, mbs: None, pass: false }), + } + } + while b.recent.len() > RECENT { + b.recent.pop_front(); + } + b.testing = candidates.iter().map(|(ip, _, _)| *ip).collect(); + } + + // Confirm promising IPs solo, fastest screen result first. + candidates.sort_by(|a, b| b.2.partial_cmp(&a.2).unwrap_or(std::cmp::Ordering::Equal)); + for (ip, lat, _) in candidates { + if shared.lock().unwrap().found { + break 'consume; + } + let r = speed::measure_download(node.as_ref(), ip, secs(SPEED_TIMEOUT), SPEED_BYTES).await; + let (mbs, pass) = match &r { + Ok(s) => (Some(*s), *s >= target), + Err(_) => (None, false), + }; + let mut b = shared.lock().unwrap(); + b.testing.retain(|x| *x != ip); + b.recent.push_back(Finished { ip, mbs, pass }); + while b.recent.len() > RECENT { + b.recent.pop_front(); + } + if pass { + b.found = true; + hit = Some((ip, lat, mbs.unwrap())); + break 'consume; + } + } + } + + stop.store(true, Ordering::Relaxed); + let _ = ui.await; // render loop clears the view before returning + producer.abort(); + + match hit { + Some((ip, latency, mbs)) => { + let alias = format!("{mbs:.2}MB-{:.0}ms-{}", latency.as_secs_f64() * 1000.0, ip); + let url = node.to_url(ip, &alias); + print_easy_found(ip, mbs, latency, &url); + write_file(&args.output.join("result.txt"), &format!("{url}\n"))?; + } + None => eprintln!( + "{} no IP reached {target:.2} MB/s after testing {tested} valid IP(s)", + style("✗").red().bold() + ), + } + Ok(()) +} + +/// Background producer: ping → latency to mint validated IPs into the bucket, +/// blocking while the bucket is full, until `max_valid` are minted, the +/// consumer signals `found`, or the source dries up. +async fn produce( + ranges: Arc, + node: Arc, + max_valid: usize, + ipv6: bool, + shared: Shared, +) { + let mut barren = 0usize; + 'outer: loop { + { + let b = shared.lock().unwrap(); + if b.found || b.valid >= max_valid { + break; + } + } + let cfg = ping::PingConfig { + count: POOL, + timeout: secs(PING_TIMEOUT), + concurrency: PING_CONCURRENCY, + ipv6, + port: 443, + max_probe: POOL.saturating_mul(200), + times: 1, + min_latency: millis(PING_MIN_MS), + max_latency: millis(PING_MAX_MS), + max_loss: 0.0, + }; + let reachable = match ping::run(ranges.as_ref(), &cfg, |_| {}).await { + Ok(r) => r, + Err(_) => break, + }; + if reachable.is_empty() { + barren += 1; + if barren >= MAX_BARREN_ROUNDS { + break; + } + continue; + } + let ips: Vec = reachable.iter().map(|r| r.ip).collect(); + let passed = latency::run( + node.as_ref(), + &ips, + LAT_CONCURRENCY, + secs(LAT_TIMEOUT), + millis(LAT_MAX_MS), + 0, + |_| {}, + ) + .await; + + let mut minted = 0usize; + for r in passed { + if !enqueue(&shared, (r.ip, r.latency), max_valid).await { + break 'outer; // found, or hit the valid cap + } + minted += 1; + } + if minted == 0 { + barren += 1; + if barren >= MAX_BARREN_ROUNDS { + break; + } + } else { + barren = 0; + } + } + shared.lock().unwrap().producer_done = true; +} + +/// Wait for a free bucket slot and enqueue `item`; return false if we should +/// stop minting (consumer found a hit, or the valid cap is reached). +async fn enqueue(shared: &Shared, item: (IpAddr, Duration), max_valid: usize) -> bool { + loop { + { + let mut b = shared.lock().unwrap(); + if b.found || b.valid >= max_valid { + return false; + } + if b.waiting.len() + b.testing.len() < POOL { + b.waiting.push_back(item); + b.valid += 1; + return true; + } + } + tokio::time::sleep(POLL).await; + } +} + +/// Periodically redraw the bucket view until `stop`, then clear it. +async fn render_loop(shared: Shared, view: BucketView, target: f64, max: usize, stop: Arc) { + let mut tick = 0usize; + loop { + view.render(&shared, target, max, tick); + if stop.load(Ordering::Relaxed) { + break; + } + tick += 1; + tokio::time::sleep(FRAME).await; + } + view.clear(); +} + +/// The fixed, in-place "bucket / recent" panel: a header line, the bucket rows, +/// and the recent-results rows, each an indicatif line whose message we set. +struct BucketView { + _mp: MultiProgress, + header: ProgressBar, + bucket_label: ProgressBar, + bucket: Vec, + recent_label: ProgressBar, + recent: Vec, +} + +impl BucketView { + fn new() -> Self { + let mp = MultiProgress::new(); + let line = |mp: &MultiProgress| { + let pb = mp.add(ProgressBar::new(1)); + pb.set_style(ProgressStyle::with_template("{msg}").unwrap()); + pb + }; + let header = line(&mp); + let bucket_label = line(&mp); + let bucket = (0..POOL).map(|_| line(&mp)).collect(); + let recent_label = line(&mp); + let recent = (0..RECENT).map(|_| line(&mp)).collect(); + bucket_label.set_message(format!(" {}", style("bucket").dim())); + recent_label.set_message(format!(" {}", style("recent").dim())); + Self { _mp: mp, header, bucket_label, bucket, recent_label, recent } + } + + fn render(&self, shared: &Shared, target: f64, max: usize, tick: usize) { + let frame = FRAMES[tick % FRAMES.len()]; + let (testing, waiting, recent, valid) = { + let b = shared.lock().unwrap(); + ( + b.testing.clone(), + b.waiting.iter().map(|(ip, _)| *ip).collect::>(), + b.recent + .iter() + .rev() // newest first + .map(|f| (f.ip, f.mbs, f.pass)) + .collect::>(), + b.valid, + ) + }; + + self.header.set_message(format!( + "{} {}/{} valid target ≥ {:.2} MB/s", + style(frame).cyan(), + valid, + max, + target + )); + + // Bucket rows: the batch under test (spinner), then the ones waiting. + let mut rows: Vec = Vec::new(); + for ip in &testing { + rows.push(format!( + " {} {:<15} {}", + style(frame).cyan(), + ip.to_string(), + style("testing…").cyan() + )); + } + for ip in &waiting { + rows.push(format!(" {:<15} {}", ip.to_string(), style("waiting").dim())); + } + for (pb, row) in self.bucket.iter().zip(rows.iter().chain(std::iter::repeat(&String::new()))) { + pb.set_message(row.clone()); + } + + // Recent results, newest first. + for (i, pb) in self.recent.iter().enumerate() { + match recent.get(i) { + Some((ip, mbs, pass)) => { + let mark = if *pass { style("✓").green() } else { style("✗").red() }; + let value = match mbs { + Some(s) => format!("{s:.2} MB/s"), + None => "failed".to_string(), + }; + let value = if *pass { style(value).green() } else { style(value).dim() }; + pb.set_message(format!(" {mark} {:<15} {value}", ip.to_string())); + } + None => pb.set_message(String::new()), + } + } + } + + fn clear(&self) { + for pb in std::iter::once(&self.header) + .chain(std::iter::once(&self.bucket_label)) + .chain(self.bucket.iter()) + .chain(std::iter::once(&self.recent_label)) + .chain(self.recent.iter()) + { + pb.finish_and_clear(); + } + } +} + +fn spin(msg: &str) -> ProgressBar { + let pb = ProgressBar::new_spinner(); + pb.enable_steady_tick(Duration::from_millis(90)); + pb.set_message(msg.to_string()); + pb +} + +fn secs(s: f64) -> Duration { + Duration::from_secs_f64(s) +} + +fn millis(ms: f64) -> Duration { + Duration::from_secs_f64(ms / 1000.0) +} diff --git a/src/main.rs b/src/main.rs index 63e6527..9278305 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,6 +9,7 @@ mod cli; mod cloudflare; mod commands; +mod easy; mod latency; mod ping; mod report; @@ -31,6 +32,7 @@ async fn main() -> Result<()> { Some(Command::Latency(args)) => commands::run_latency(args).await, Some(Command::Speed(args)) => commands::run_speed(args).await, Some(Command::Export(args)) => commands::run_export(args), + Some(Command::Easy(args)) => easy::run(args).await, None => commands::run_auto(cli.auto).await, } } diff --git a/src/ping.rs b/src/ping.rs index 46849c7..c2ab4a8 100644 --- a/src/ping.rs +++ b/src/ping.rs @@ -19,7 +19,8 @@ use std::time::{Duration, Instant}; use anyhow::Result; use futures::stream::{FuturesUnordered, StreamExt}; use ipnet::IpNet; -use rand::Rng; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; use tokio::net::TcpStream; use tokio::time::timeout; @@ -132,7 +133,7 @@ pub async fn run( let mut attempts = vec![0u32; nets.len()]; let mut hits = vec![0u32; nets.len()]; let mut seen: HashSet = HashSet::new(); - let mut rng = rand::thread_rng(); + let mut rng = StdRng::from_entropy(); // Keep `concurrency` probes in flight, refilling one as each finishes so the // weights learned so far steer every new draw. diff --git a/src/report.rs b/src/report.rs index 82f7a48..0040a1b 100644 --- a/src/report.rs +++ b/src/report.rs @@ -180,7 +180,7 @@ pub(crate) fn print_table(results: &[T]) { eprintln!(); eprintln!("{}", style(format!(" Fastest {} IPs", results.len())).bold().underlined()); let ip_w = ip_col_width(results.iter().map(|r| r.ip())); - eprintln!(" {:>3} {:3} {:3} {:3} {:(results: &[T], path: &Path) -> Result<()> { let mut body = String::with_capacity(results.len() * 16); diff --git a/src/speed.rs b/src/speed.rs index c667b5a..598cd8a 100644 --- a/src/speed.rs +++ b/src/speed.rs @@ -1,13 +1,14 @@ //! Stage 3 (`speed`): download through the node via each candidate IP and //! measure real throughput, then rank by speed bucket then latency. -use std::net::IpAddr; +use std::net::{IpAddr, SocketAddr}; use std::time::{Duration, Instant}; -use anyhow::{Result, anyhow}; +use anyhow::{Context, Result, anyhow}; use futures::stream::{self, StreamExt}; use rustls::pki_types::ServerName; use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpStream, lookup_host}; use crate::vless::{self, VlessNode}; @@ -15,6 +16,14 @@ const TARGET_HOST: &str = "cachefly.cachefly.net"; const TARGET_PORT: u16 = 443; const TARGET_PATH: &str = "/50mb.test"; +// Direct baseline target: a domestic, no-proxy-reachable speedtest file. The +// proxied path is bounded by the local link, and abroad hosts are unreachable +// without a proxy here, so we gauge the link ceiling against a CN mirror. TUNA's +// /speedtest/ files are large (1GB) and stable; swap if it ever moves. +const DIRECT_HOST: &str = "mirrors.tuna.tsinghua.edu.cn"; +const DIRECT_PORT: u16 = 443; +const DIRECT_PATH: &str = "/speedtest/1000mb.bin"; + /// One IP with its (carried-over) latency and measured download speed (MB/s). #[derive(Clone, Debug)] pub struct SpeedResult { @@ -92,12 +101,57 @@ pub async fn measure_download( .await .map_err(|_| anyhow!("connect timeout"))??; let sni = ServerName::try_from(TARGET_HOST.to_string())?; - let mut stream = tokio::time::timeout(timeout, vless::tls_connector().connect(sni, tunnel)) + let stream = tokio::time::timeout(timeout, vless::tls_connector().connect(sni, tunnel)) .await .map_err(|_| anyhow!("tls timeout"))??; + download_speed(stream, TARGET_HOST, TARGET_PATH, timeout, limit_bytes).await +} +/// Baseline download with no proxy: connect straight to a domestic test host +/// and measure. A proxied node can never beat the local link, so this bounds +/// what the `easy` command can sensibly target. +pub async fn measure_direct(timeout: Duration, limit_bytes: u64) -> Result { + let addr = resolve_preferring_v4(DIRECT_HOST, DIRECT_PORT).await?; + let tcp = tokio::time::timeout(timeout, TcpStream::connect(addr)) + .await + .map_err(|_| anyhow!("connect timeout"))??; + tcp.set_nodelay(true).ok(); + let sni = ServerName::try_from(DIRECT_HOST.to_string())?; + let stream = tokio::time::timeout(timeout, vless::tls_connector().connect(sni, tcp)) + .await + .map_err(|_| anyhow!("tls timeout"))??; + download_speed(stream, DIRECT_HOST, DIRECT_PATH, timeout, limit_bytes).await +} + +/// Resolve `host:port`, preferring an IPv4 address. Networks this tool targets +/// often have broken IPv6; `TcpStream::connect((host, port))` would try a dead +/// AAAA first and stall, so we resolve and pick the address ourselves. +async fn resolve_preferring_v4(host: &str, port: u16) -> Result { + let addrs: Vec = + lookup_host((host, port)).await.context("resolve host")?.collect(); + addrs + .iter() + .copied() + .find(SocketAddr::is_ipv4) + .or_else(|| addrs.first().copied()) + .ok_or_else(|| anyhow!("no address for {host}")) +} + +/// GET the test file over an already-connected TLS stream and measure +/// throughput. Stops at steady state, the byte cap, or the time cap — whichever +/// comes first. Shared by the proxied and direct paths. +async fn download_speed( + mut stream: S, + host: &str, + path: &str, + timeout: Duration, + limit_bytes: u64, +) -> Result +where + S: AsyncReadExt + AsyncWriteExt + Unpin, +{ let request = format!( - "GET {TARGET_PATH} HTTP/1.1\r\nHost: {TARGET_HOST}\r\n\ + "GET {path} HTTP/1.1\r\nHost: {host}\r\n\ User-Agent: fast-xray\r\nAccept: */*\r\nConnection: close\r\n\r\n" ); stream.write_all(request.as_bytes()).await?;