astelm
/
telemt
mirror of https://github.com/telemt/telemt.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Revert "Skip IPv6 connections when IPv6 is unavailable on host" 

This reverts commit d122cbfe5a.
This commit is contained in:
Igor 2026-02-18 08:56:42 +03:00
parent d122cbfe5a
commit 496a4ab005
4 changed files with 85 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/main.rs
View File

@ -35,7 +35,7 @@ use crate::transport::middle_proxy::{
stun_probe, stun_probe,
}; };
use crate::transport::{ListenOptions, UpstreamManager, create_listener}; use crate::transport::{ListenOptions, UpstreamManager, create_listener};
use crate::util::ip::{detect_ip, check_ipv6_available}; use crate::util::ip::detect_ip;
use crate::protocol::constants::{TG_MIDDLE_PROXIES_V4, TG_MIDDLE_PROXIES_V6}; use crate::protocol::constants::{TG_MIDDLE_PROXIES_V4, TG_MIDDLE_PROXIES_V6};
fn parse_cli() -> (String, bool, Option<String>) { fn parse_cli() -> (String, bool, Option<String>) {
@ -219,19 +219,7 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
warn!("Using default tls_domain. Consider setting a custom domain."); warn!("Using default tls_domain. Consider setting a custom domain.");
} }
let ipv6_available = check_ipv6_available(); let prefer_ipv6 = config.general.prefer_ipv6;
if ipv6_available {
info!("IPv6: available");
} else {
warn!("IPv6: not available on this host");
}
let prefer_ipv6 = if config.general.prefer_ipv6 && !ipv6_available {
warn!("prefer_ipv6 is set but IPv6 is not available, falling back to IPv4");
false
} else {
config.general.prefer_ipv6
};
let mut use_middle_proxy = config.general.use_middle_proxy; let mut use_middle_proxy = config.general.use_middle_proxy;
let config = Arc::new(config); let config = Arc::new(config);
let stats = Arc::new(Stats::new()); let stats = Arc::new(Stats::new());
@ -354,12 +342,6 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
cfg_v6.map = crate::protocol::constants::TG_MIDDLE_PROXIES_V6.clone(); cfg_v6.map = crate::protocol::constants::TG_MIDDLE_PROXIES_V6.clone();
} }
let effective_v6_map = if ipv6_available {
cfg_v6.map.clone()
} else {
std::collections::HashMap::new()
};
let pool = MePool::new( let pool = MePool::new(
proxy_tag, proxy_tag,
proxy_secret, proxy_secret,
@ -367,7 +349,7 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
config.general.middle_proxy_nat_probe, config.general.middle_proxy_nat_probe,
config.general.middle_proxy_nat_stun.clone(), config.general.middle_proxy_nat_stun.clone(),
cfg_v4.map.clone(), cfg_v4.map.clone(),
effective_v6_map, cfg_v6.map.clone(),
cfg_v4.default_dc.or(cfg_v6.default_dc), cfg_v4.default_dc.or(cfg_v6.default_dc),
); );
@ -380,7 +362,7 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
let rng_clone = rng.clone(); let rng_clone = rng.clone();
tokio::spawn(async move { tokio::spawn(async move {
crate::transport::middle_proxy::me_health_monitor( crate::transport::middle_proxy::me_health_monitor(
pool_clone, rng_clone, 2, ipv6_available, pool_clone, rng_clone, 2,
) )
.await; .await;
}); });
@ -500,7 +482,7 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
info!("================= Telegram DC Connectivity ================="); info!("================= Telegram DC Connectivity =================");
let ping_results = upstream_manager let ping_results = upstream_manager
.ping_all_dcs(prefer_ipv6, &config.dc_overrides, ipv6_available) .ping_all_dcs(prefer_ipv6, &config.dc_overrides)
.await; .await;
for upstream_result in &ping_results { for upstream_result in &ping_results {
@ -578,7 +560,7 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
// Background tasks // Background tasks
let um_clone = upstream_manager.clone(); let um_clone = upstream_manager.clone();
tokio::spawn(async move { tokio::spawn(async move {
um_clone.run_health_checks(prefer_ipv6, ipv6_available).await; um_clone.run_health_checks(prefer_ipv6).await;
}); });
let rc_clone = replay_checker.clone(); let rc_clone = replay_checker.clone();

7
src/transport/middle_proxy/health.rs
View File

@ -10,7 +10,7 @@ use crate::crypto::SecureRandom;
use super::MePool; use super::MePool;
pub async fn me_health_monitor(pool: Arc<MePool>, rng: Arc<SecureRandom>, _min_connections: usize, ipv6_available: bool) { pub async fn me_health_monitor(pool: Arc<MePool>, rng: Arc<SecureRandom>, _min_connections: usize) {
let mut backoff: HashMap<i32, u64> = HashMap::new(); let mut backoff: HashMap<i32, u64> = HashMap::new();
let mut last_attempt: HashMap<i32, Instant> = HashMap::new(); let mut last_attempt: HashMap<i32, Instant> = HashMap::new();
loop { loop {
@ -63,10 +63,7 @@ pub async fn me_health_monitor(pool: Arc<MePool>, rng: Arc<SecureRandom>, _min_c
} }
} }
// IPv6 coverage check (skip if IPv6 not available on host) // IPv6 coverage check (if available)
if !ipv6_available {
continue;
}
let map_v6 = pool.proxy_map_v6.read().await.clone(); let map_v6 = pool.proxy_map_v6.read().await.clone();
let writer_addrs_v6: std::collections::HashSet<SocketAddr> = pool let writer_addrs_v6: std::collections::HashSet<SocketAddr> = pool
.writers .writers

178
src/transport/upstream.rs
View File

@ -355,7 +355,6 @@ impl UpstreamManager {
&self, &self,
prefer_ipv6: bool, prefer_ipv6: bool,
dc_overrides: &HashMap<String, Vec<String>>, dc_overrides: &HashMap<String, Vec<String>>,
ipv6_available: bool,
) -> Vec<StartupPingResult> { ) -> Vec<StartupPingResult> {
let upstreams: Vec<(usize, UpstreamConfig)> = { let upstreams: Vec<(usize, UpstreamConfig)> = {
let guard = self.upstreams.read().await; let guard = self.upstreams.read().await;
@ -378,45 +377,43 @@ impl UpstreamManager {
let mut v6_results = Vec::new(); let mut v6_results = Vec::new();
let mut v4_results = Vec::new(); let mut v4_results = Vec::new();
// === Ping IPv6 first (skip if IPv6 not available) === // === Ping IPv6 first ===
if ipv6_available { for dc_zero_idx in 0..NUM_DCS {
for dc_zero_idx in 0..NUM_DCS { let dc_v6 = TG_DATACENTERS_V6[dc_zero_idx];
let dc_v6 = TG_DATACENTERS_V6[dc_zero_idx]; let addr_v6 = SocketAddr::new(dc_v6, TG_DATACENTER_PORT);
let addr_v6 = SocketAddr::new(dc_v6, TG_DATACENTER_PORT);
let result = tokio::time::timeout( let result = tokio::time::timeout(
Duration::from_secs(DC_PING_TIMEOUT_SECS), Duration::from_secs(DC_PING_TIMEOUT_SECS),
self.ping_single_dc(&upstream_config, addr_v6) self.ping_single_dc(&upstream_config, addr_v6)
).await; ).await;
let ping_result = match result { let ping_result = match result {
Ok(Ok(rtt_ms)) => { Ok(Ok(rtt_ms)) => {
let mut guard = self.upstreams.write().await; let mut guard = self.upstreams.write().await;
if let Some(u) = guard.get_mut(*upstream_idx) { if let Some(u) = guard.get_mut(*upstream_idx) {
u.dc_latency[dc_zero_idx].update(rtt_ms); u.dc_latency[dc_zero_idx].update(rtt_ms);
}
DcPingResult {
dc_idx: dc_zero_idx + 1,
dc_addr: addr_v6,
rtt_ms: Some(rtt_ms),
error: None,
}
} }
Ok(Err(e)) => DcPingResult { DcPingResult {
dc_idx: dc_zero_idx + 1, dc_idx: dc_zero_idx + 1,
dc_addr: addr_v6, dc_addr: addr_v6,
rtt_ms: None, rtt_ms: Some(rtt_ms),
error: Some(e.to_string()), error: None,
}, }
Err(_) => DcPingResult { }
dc_idx: dc_zero_idx + 1, Ok(Err(e)) => DcPingResult {
dc_addr: addr_v6, dc_idx: dc_zero_idx + 1,
rtt_ms: None, dc_addr: addr_v6,
error: Some("timeout".to_string()), rtt_ms: None,
}, error: Some(e.to_string()),
}; },
v6_results.push(ping_result); Err(_) => DcPingResult {
} dc_idx: dc_zero_idx + 1,
dc_addr: addr_v6,
rtt_ms: None,
error: Some("timeout".to_string()),
},
};
v6_results.push(ping_result);
} }
// === Then ping IPv4 === // === Then ping IPv4 ===
@ -520,12 +517,8 @@ impl UpstreamManager {
let mut guard = self.upstreams.write().await; let mut guard = self.upstreams.write().await;
if let Some(u) = guard.get_mut(*upstream_idx) { if let Some(u) = guard.get_mut(*upstream_idx) {
for dc_zero_idx in 0..NUM_DCS { for dc_zero_idx in 0..NUM_DCS {
let v6_ok = v6_results.get(dc_zero_idx) let v6_ok = v6_results[dc_zero_idx].rtt_ms.is_some();
.map(|r| r.rtt_ms.is_some()) let v4_ok = v4_results[dc_zero_idx].rtt_ms.is_some();
.unwrap_or(false);
let v4_ok = v4_results.get(dc_zero_idx)
.map(|r| r.rtt_ms.is_some())
.unwrap_or(false);
u.dc_ip_pref[dc_zero_idx] = match (v6_ok, v4_ok) { u.dc_ip_pref[dc_zero_idx] = match (v6_ok, v4_ok) {
(true, true) => IpPreference::BothWork, (true, true) => IpPreference::BothWork,
@ -558,7 +551,7 @@ impl UpstreamManager {
/// Background health check: rotates through DCs, 30s interval. /// Background health check: rotates through DCs, 30s interval.
/// Uses preferred IP version based on config. /// Uses preferred IP version based on config.
pub async fn run_health_checks(&self, prefer_ipv6: bool, ipv6_available: bool) { pub async fn run_health_checks(&self, prefer_ipv6: bool) {
let mut dc_rotation = 0usize; let mut dc_rotation = 0usize;
loop { loop {
@ -567,19 +560,16 @@ impl UpstreamManager {
let dc_zero_idx = dc_rotation % NUM_DCS; let dc_zero_idx = dc_rotation % NUM_DCS;
dc_rotation += 1; dc_rotation += 1;
let dc_addr = if prefer_ipv6 && ipv6_available { let dc_addr = if prefer_ipv6 {
SocketAddr::new(TG_DATACENTERS_V6[dc_zero_idx], TG_DATACENTER_PORT) SocketAddr::new(TG_DATACENTERS_V6[dc_zero_idx], TG_DATACENTER_PORT)
} else { } else {
SocketAddr::new(TG_DATACENTERS_V4[dc_zero_idx], TG_DATACENTER_PORT) SocketAddr::new(TG_DATACENTERS_V4[dc_zero_idx], TG_DATACENTER_PORT)
}; };
// Skip IPv6 fallback if IPv6 is not available let fallback_addr = if prefer_ipv6 {
let fallback_addr = if !ipv6_available { SocketAddr::new(TG_DATACENTERS_V4[dc_zero_idx], TG_DATACENTER_PORT)
None
} else if prefer_ipv6 {
Some(SocketAddr::new(TG_DATACENTERS_V4[dc_zero_idx], TG_DATACENTER_PORT))
} else { } else {
Some(SocketAddr::new(TG_DATACENTERS_V6[dc_zero_idx], TG_DATACENTER_PORT)) SocketAddr::new(TG_DATACENTERS_V6[dc_zero_idx], TG_DATACENTER_PORT)
}; };
let count = self.upstreams.read().await.len(); let count = self.upstreams.read().await.len();
@ -615,67 +605,53 @@ impl UpstreamManager {
u.last_check = std::time::Instant::now(); u.last_check = std::time::Instant::now();
} }
Ok(Err(_)) | Err(_) => { Ok(Err(_)) | Err(_) => {
// Try fallback (only if fallback address is available) // Try fallback
if let Some(fb_addr) = fallback_addr { debug!(dc = dc_zero_idx + 1, "Health check failed, trying fallback");
debug!(dc = dc_zero_idx + 1, "Health check failed, trying fallback");
let start2 = Instant::now(); let start2 = Instant::now();
let result2 = tokio::time::timeout( let result2 = tokio::time::timeout(
Duration::from_secs(10), Duration::from_secs(10),
self.connect_via_upstream(&config, fb_addr) self.connect_via_upstream(&config, fallback_addr)
).await; ).await;
let mut guard = self.upstreams.write().await; let mut guard = self.upstreams.write().await;
let u = &mut guard[i]; let u = &mut guard[i];
match result2 { match result2 {
Ok(Ok(_stream)) => { Ok(Ok(_stream)) => {
let rtt_ms = start2.elapsed().as_secs_f64() * 1000.0; let rtt_ms = start2.elapsed().as_secs_f64() * 1000.0;
u.dc_latency[dc_zero_idx].update(rtt_ms); u.dc_latency[dc_zero_idx].update(rtt_ms);
if !u.healthy { if !u.healthy {
info!( info!(
rtt = format!("{:.0} ms", rtt_ms), rtt = format!("{:.0} ms", rtt_ms),
dc = dc_zero_idx + 1, dc = dc_zero_idx + 1,
"Upstream recovered (fallback)" "Upstream recovered (fallback)"
); );
}
u.healthy = true;
u.fails = 0;
} }
Ok(Err(e)) => { u.healthy = true;
u.fails += 1; u.fails = 0;
debug!(dc = dc_zero_idx + 1, fails = u.fails, }
"Health check failed (both): {}", e); Ok(Err(e)) => {
if u.fails > 3 { u.fails += 1;
u.healthy = false; debug!(dc = dc_zero_idx + 1, fails = u.fails,
warn!("Upstream unhealthy (fails)"); "Health check failed (both): {}", e);
} if u.fails > 3 {
} u.healthy = false;
Err(_) => { warn!("Upstream unhealthy (fails)");
u.fails += 1;
debug!(dc = dc_zero_idx + 1, fails = u.fails,
"Health check timeout (both)");
if u.fails > 3 {
u.healthy = false;
warn!("Upstream unhealthy (timeout)");
}
} }
} }
u.last_check = std::time::Instant::now(); Err(_) => {
} else { u.fails += 1;
// No fallback available, mark failure directly debug!(dc = dc_zero_idx + 1, fails = u.fails,
let mut guard = self.upstreams.write().await; "Health check timeout (both)");
let u = &mut guard[i]; if u.fails > 3 {
u.fails += 1; u.healthy = false;
debug!(dc = dc_zero_idx + 1, fails = u.fails, warn!("Upstream unhealthy (timeout)");
"Health check failed (no fallback)"); }
if u.fails > 3 {
u.healthy = false;
warn!("Upstream unhealthy (fails)");
} }
u.last_check = std::time::Instant::now();
} }
u.last_check = std::time::Instant::now();
} }
} }
} }

6
src/util/ip.rs
View File

@ -54,12 +54,6 @@ fn get_local_ipv6(target: &str) -> Option<IpAddr> {
socket.local_addr().ok().map(|addr| addr.ip()) socket.local_addr().ok().map(|addr| addr.ip())
} }
/// Check if IPv6 connectivity is available on this host.
/// Uses UDP connect to Google DNS IPv6 (no packets sent).
pub fn check_ipv6_available() -> bool {
get_local_ipv6("[2001:4860:4860::8888]:80").is_some()
}
/// Detect public IP addresses /// Detect public IP addresses
pub async fn detect_ip() -> IpInfo { pub async fn detect_ip() -> IpInfo {
let mut info = IpInfo::default(); let mut info = IpInfo::default();