Recover managed TUN workers when the underlying network changes

Monitor the selected IPv4 and IPv6 native default paths, including brief link outages. Retain the owned TUN, routing/firewall lease, redirect ports and fixed TCP listeners while retiring network-dependent work across reload generations.

Rebuild DNS, proxy transports, probes and WireGuard devices from the accepted configuration. Preserve fake-IP mappings and explicit selections, pause offline, retry failed startup, and allow a reload to repair listeners that cannot restart. Keep unmanaged deployments outside task-lifetime tracking.

Cover task retirement, QUIC driver completion, resolver/probe rebuilding, process recovery and reload repair, plus isolated dual-stack kernel handover with REDIRECT on and off. Document recovery limits in both TUN guides.
This commit is contained in:
cfal
2026-11-06 07:57:45 +00:00
parent 3965230d31
commit d9c62e241e
56 changed files with 2655 additions and 204 deletions
+27 -2
View File
@@ -228,13 +228,38 @@ mark-aware reverse-path filtering.
No universal VPN, firewall, resolver-manager, or distribution compatibility is
promised.
Reload retains the interface, capture, and redirect listener identities. Network
topology changes require restart and are rejected before retiring live workers.
Reload retains the interface, capture, and redirect listener identities. Managed
TUN topology configuration changes require restart and are rejected before retiring live workers.
Ordinary worker reloads can reset existing connections and briefly interrupt
traffic, but do not disable capture. Failed candidates roll back on the same
network lease. Unexpected owned routing changes or a missing private firewall
table cause orderly shutdown. Resolver-setting changes do not.
After startup, changes to the underlying main-table default path, gateway, or
explicit preferred source, and loss of the selected source address, trigger recovery
in place. Ordinary IPv6 privacy-address rotation does not reset existing flows.
Both IP families are monitored. A
complete outage pauses forwarding until an underlay returns; temporary recovery
failures are retried without removing capture. Recovery rebuilds DNS and proxy
transports from the accepted in-memory configuration, not from edited files.
Explicit reloads wait while offline and can replace a configuration that prevents
workers from restarting once an underlay is available.
The TUN, REDIRECT ports, fake-IP mappings, and explicit selector choices remain
owned and retained. WireGuard peer hostnames are refreshed best-effort; a failed
refresh retains the previous endpoints rather than blocking unrelated workers. Old flows,
pending connections, and connections draining from earlier reloads are cancelled.
Applications must reconnect; requests are not replayed and existing TCP/QUIC
sessions are not migrated. TUN TCP resets are best-effort.
Detection uses netlink notifications and periodic checks, with a short settling
delay. It observes kernel route/source selection for representative off-link
destinations, not Internet reachability or every destination-specific policy.
Packet loss, captive portals, or a Wi-Fi change that preserves the observed
routing and carrier state may not trigger recovery. Explicit `bind_interface`
and listen-address settings remain fixed; they are never silently moved to another
interface. A reset affects all proxy workers in this process, including local
listeners.
shoes exclusively owns its token-named `inet` nftables table, including all of its
contents. Do not add rules, chains, or sets to that table: external changes inside
it are neither audited nor preserved. Cleanup removes the exact owned table even
+5 -1
View File
@@ -127,7 +127,11 @@ TCP REDIRECT 要求每个加速地址族都有可用的环回网络。如果启
冲突的表、优先级(包括固定的后备优先级 `32768`)、标记保留或前面的非本地策略规则会导致启动拒绝。其他 VPN 和防火墙管理器需要明确的共存验证; shoes 不会覆盖外部防火墙的决定。 REDIRECT 还拒绝使用其虚拟对等点作为网关的本机路由和不透明的下一跳对象路由。捕获的回复在标记感知反向路径过滤之前携带捕获标记。不承诺通用 VPN、防火墙、解析器管理器或分发兼容性。
重新加载会保留接口、捕获和重定向监听器身份。网络拓扑更改需要重新启动,并且在退休工作人员之前会被拒绝。普通工作线程重新加载可以重置现有连接并短暂中断流量,但不会禁用捕获。失败的候选者将回滚同一网络租约。意外的所属路由更改或丢失的专用防火墙表会导致有序关闭。解析器设置更改则不会。
重新加载会保留接口、捕获和重定向监听器身份。托管 TUN 拓扑配置更改需要重新启动,并且在停止现有工作线程之前会被拒绝。普通工作线程重新加载可以重置现有连接并短暂中断流量,但不会禁用捕获。失败的候选者将回滚同一网络租约。意外的所属路由更改或丢失的专用防火墙表会导致有序关闭。解析器设置更改则不会。
启动后,底层主路由表的默认路径、网关或显式首选源地址发生变化,或当前选用的源地址被删除时,shoes 会在进程内恢复连接。普通 IPv6 隐私地址轮换不会重置现有连接。IPv4 和 IPv6 分别受到监视。全部底层连接中断时暂停转发,连接恢复后重新启动工作线程;临时恢复失败会自动重试,不会移除捕获规则。恢复使用已接受的内存配置重建 DNS 和代理传输,不会重新读取已编辑的配置文件。显式重载在离线时等待,底层网络可用后可替换导致工作线程无法重启的配置。TUN、REDIRECT 端口、fake-IP 映射和显式选择的代理保持不变。WireGuard 对等端主机名会尽力重新解析;失败时保留原有端点,不阻塞其他工作线程。旧连接、正在建立的连接,以及此前重新加载后仍在排空的连接都会被取消。应用程序必须重新连接;shoes 不会重放请求,也不会迁移现有 TCP/QUIC 会话。TUN TCP 重置为尽力而为。
检测使用 netlink 通知和周期检查,并等待短暂的稳定时间。它观察内核对代表性非直连目标的路由和源地址选择,不检测互联网可达性,也不覆盖所有按目标配置的策略路由。丢包、认证门户或保持相同路由和载波状态的 Wi-Fi 切换可能不会触发恢复。显式 `bind_interface` 和监听地址始终保持固定,不会自动移到其他接口。重置影响本进程中的所有代理工作线程,包括本地监听器。
shoes 独家拥有其名为 `inet` nftables 表的代币,包括其所有内容。不要向该表添加规则、链或集:其中的外部更改既不会被审核也不会被保留。清理会删除确切拥有的表,即使其内容已更改;其他表保持不变。连接网络集通过 nftables 自动更新并缓存在内存中,而不是记录为防火墙快照。
+11 -12
View File
@@ -80,10 +80,10 @@ impl SessionPool {
) {
// Each checkout opens exactly one stream; only its idempotent terminal
// transition may release it. Idle sessions have no active streams.
let Ok(runtime) = tokio::runtime::Handle::try_current() else {
if tokio::runtime::Handle::try_current().is_err() {
session.close();
return;
};
}
if self.options.disable_reuse || !session.is_reusable() {
session.close();
return;
@@ -98,17 +98,16 @@ impl SessionPool {
let weak = Arc::downgrade(self);
let check_interval = interval(self.options.idle_session_check_interval_secs);
state.reaper = Some(
runtime
.spawn(async move {
loop {
tokio::time::sleep(check_interval).await;
let Some(pool) = weak.upgrade() else { return };
if !pool.reap() {
return;
}
crate::network_tasks::spawn(async move {
loop {
tokio::time::sleep(check_interval).await;
let Some(pool) = weak.upgrade() else { return };
if !pool.reap() {
return;
}
})
.abort_handle(),
}
})
.abort_handle(),
);
}
}
+3 -3
View File
@@ -327,7 +327,7 @@ impl ClientSessionState {
W: tokio::io::AsyncWrite + Send + Unpin + 'static,
{
let writer_state = Arc::clone(&session);
let writer_task = tokio::spawn(async move {
let writer_task = crate::network_tasks::spawn(async move {
tokio::select! {
result = Self::writer_loop(
Arc::downgrade(&writer_state), writer, outgoing_rx,
@@ -341,8 +341,8 @@ impl ClientSessionState {
}
writer_state.close();
});
let watchdog_task = tokio::spawn(Self::watchdog_loop(Arc::clone(&session)));
let reader_task = tokio::spawn(async move {
let watchdog_task = crate::network_tasks::spawn(Self::watchdog_loop(Arc::clone(&session)));
let reader_task = crate::network_tasks::spawn(async move {
tokio::select! {
result = Self::reader_loop(
Arc::downgrade(&session), reader, session.close_notify.clone(),
+1 -1
View File
@@ -439,7 +439,7 @@ impl AnyTlsSession {
if let Some(stream) = stream_opt {
let session = Arc::clone(self);
let handle = tokio::spawn(async move {
let handle = crate::network_tasks::spawn(async move {
let _permit = permit;
if let Err(e) = session.handle_new_stream(stream).await {
log::debug!("AnyTLS stream {} error: {}", stream_id, e);
+47 -10
View File
@@ -42,11 +42,16 @@ struct FullBudgetAttempts {
}
pub struct ProbeState {
connectors: Mutex<HashMap<String, Arc<ClientChainGroup>>>,
transports: Mutex<Arc<ProbeTransports>>,
history: Mutex<HashMap<String, History>>,
full_budget_attempts: Mutex<HashMap<String, FullBudgetAttempts>>,
}
struct ProbeTransports {
connectors: Mutex<HashMap<String, Arc<ClientChainGroup>>>,
resolver: Arc<dyn Resolver>,
forbidden: Vec<ipnetwork::IpNetwork>,
network: Option<crate::network_tasks::NetworkTasks>,
}
impl ProbeState {
@@ -61,14 +66,27 @@ impl ProbeState {
pub fn new(resolver: Arc<dyn Resolver>, forbidden: Vec<ipnetwork::IpNetwork>) -> Self {
Self {
connectors: Mutex::new(HashMap::new()),
transports: Mutex::new(Arc::new(ProbeTransports {
connectors: Mutex::new(HashMap::new()),
resolver,
forbidden,
network: crate::network_tasks::NetworkTasks::current(),
})),
history: Mutex::new(HashMap::new()),
full_budget_attempts: Mutex::new(HashMap::new()),
resolver,
forbidden,
}
}
pub(crate) fn reset_network(
&self,
resolver: Arc<dyn Resolver>,
forbidden: Vec<ipnetwork::IpNetwork>,
) {
let replacement = Self::new(resolver, forbidden);
*self.transports.lock() = replacement.transports.into_inner();
self.full_budget_attempts.lock().clear();
}
pub fn decorate(&self, name: &str, object: &mut Value) {
let histories = self.history.lock();
let Some(history) = histories.get(name) else {
@@ -166,7 +184,9 @@ impl ProbeState {
}
urls.insert(query.url.to_string(), tokio::time::Instant::now());
}
}
impl ProbeTransports {
fn connector(&self, catalog: &Catalog, name: &str) -> io::Result<Arc<ClientChainGroup>> {
if let Some(connector) = self.connectors.lock().get(name) {
return Ok(connector.clone());
@@ -669,12 +689,29 @@ async fn request_with_tls_config(
query: &ProbeQuery,
tls_config: impl FnOnce() -> rustls::ClientConfig,
) -> io::Result<(u16, bool)> {
let resolver = ProbeResolver::for_query(
generation.probes.resolver.clone(),
&generation.probes.forbidden,
query,
)?;
let chain = generation.probes.connector(&generation.catalog, name)?;
let transports = generation.probes.transports.lock().clone();
let request =
request_with_transports(&transports, &generation.catalog, name, query, tls_config);
match &transports.network {
Some(network) => tokio::select! {
biased;
_ = network.cancelled.cancelled() => Err(io::Error::new(io::ErrorKind::Interrupted, "probe network retired")),
result = network.scope(request) => result,
},
None => request.await,
}
}
async fn request_with_transports(
transports: &ProbeTransports,
catalog: &Catalog,
name: &str,
query: &ProbeQuery,
tls_config: impl FnOnce() -> rustls::ClientConfig,
) -> io::Result<(u16, bool)> {
let resolver =
ProbeResolver::for_query(transports.resolver.clone(), &transports.forbidden, query)?;
let chain = transports.connector(catalog, name)?;
let started = Instant::now();
let host = query
.url
+134 -8
View File
@@ -12,6 +12,109 @@ async fn publish_for_api_probes(runtime: &ControlRuntime, generation: &Arc<Gener
.store(false, std::sync::atomic::Ordering::Release);
}
#[tokio::test]
async fn network_reset_replaces_probe_transports_without_losing_history() {
let old_network = crate::network_tasks::NetworkLifetime::default();
let old_resolver: Arc<dyn Resolver> = Arc::new(PendingAnswers::default());
let probes = old_network
.scope(async { ProbeState::new(old_resolver.clone(), vec![]) })
.await;
probes.record("DIRECT", "http://example.test/", 12, true);
let old = probes.transports.lock().clone();
let catalog = crate::config::create_server_configs(
serde_json::from_value(json!([
{"client_group":"chosen", "type":"selector", "client_chains":[
{"chain":[{"name":"direct", "protocol":{"type":"direct"}}]}
]}
]))
.unwrap(),
)
.unwrap()
.catalog
.unwrap();
let connector = old.connector(&catalog, "DIRECT").unwrap();
old_network.cancel();
old_network.wait().await;
let new_network = crate::network_tasks::NetworkLifetime::default();
let new_resolver: Arc<dyn Resolver> = Arc::new(PendingAnswers::default());
new_network
.scope(async {
probes.reset_network(new_resolver.clone(), vec!["198.18.0.0/15".parse().unwrap()]);
})
.await;
let current = probes.transports.lock().clone();
assert!(!Arc::ptr_eq(&old, &current));
assert!(Arc::ptr_eq(&current.resolver, &new_resolver));
assert!(old.network.as_ref().unwrap().cancelled.is_cancelled());
assert!(!current.network.as_ref().unwrap().cancelled.is_cancelled());
assert!(current.connectors.lock().is_empty());
assert!(Arc::ptr_eq(
&connector,
&old.connector(&catalog, "DIRECT").unwrap()
));
assert!(current.connectors.lock().is_empty());
assert!(!Arc::ptr_eq(
&connector,
&current.connector(&catalog, "DIRECT").unwrap()
));
assert_eq!(current.forbidden, ["198.18.0.0/15".parse().unwrap()]);
let mut object = json!({});
probes.decorate("DIRECT", &mut object);
assert_eq!(object["history"][0]["delay"], 12);
}
#[tokio::test]
async fn network_reset_cancels_a_connected_stalled_probe() {
tokio::time::timeout(Duration::from_secs(3), async {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let network = crate::network_tasks::NetworkLifetime::default();
let catalog = crate::config::create_server_configs(
serde_json::from_value(json!([
{"client_group":"path", "type":"selector", "client_chains":[{"chain":[
{"name":"proxy", "address":listener.local_addr().unwrap().to_string(),
"protocol":{"type":"http"}}
]}]}
]))
.unwrap(),
)
.unwrap()
.catalog
.unwrap();
let generation = network
.scope(async {
generation(
catalog,
Arc::new(Answers(vec!["192.0.2.1:80".parse().unwrap()])),
)
})
.await;
let query = query();
let request = request(&generation, "proxy", &query);
tokio::pin!(request);
let peer = async {
let (mut stream, _) = listener.accept().await.unwrap();
let mut header = Vec::new();
while !header.ends_with(b"\r\n\r\n") {
header.push(stream.read_u8().await.unwrap());
}
network.cancel();
stream
};
let stream = tokio::select! {
result = &mut request => panic!("probe completed before reset: {result:?}"),
stream = peer => stream,
};
assert_eq!(
request.await.unwrap_err().kind(),
io::ErrorKind::Interrupted
);
network.wait().await;
drop(stream);
})
.await
.unwrap();
}
#[tokio::test]
async fn automatic_probes_use_declared_chains_not_callers_and_keep_group_settings_separate() {
tokio::time::timeout(Duration::from_secs(5), async {
@@ -1086,7 +1189,7 @@ async fn probe_wrapped_errors(h2_proxy: bool) {
proxy: Box::new(
crate::tcp::proxy_connector_impl::ProxyConnectorImpl::from_config(
config,
generation.probes.resolver.clone(),
generation.probes.transports.lock().resolver.clone(),
)
.unwrap()
.unwrap(),
@@ -1096,10 +1199,16 @@ async fn probe_wrapped_errors(h2_proxy: bool) {
InitialHopEntry::Direct(socket)
};
let chain = ClientProxyChain::new(vec![entry], vec![]);
generation.probes.connectors.lock().insert(
"member-0".into(),
Arc::new(ClientChainGroup::new(vec![chain])),
);
generation
.probes
.transports
.lock()
.connectors
.lock()
.insert(
"member-0".into(),
Arc::new(ClientChainGroup::new(vec![chain])),
);
generation
.catalog
.selection
@@ -2274,8 +2383,8 @@ async fn review_forbidden_literals_do_not_construct_or_cache_probe_transports()
)),
);
let mut generation = generation(catalog, Arc::new(Answers(Vec::new())));
generation
.probes
Arc::get_mut(generation.probes.transports.get_mut())
.unwrap()
.forbidden
.push("100.64.0.0/10".parse().unwrap());
for url in [
@@ -2288,7 +2397,16 @@ async fn review_forbidden_literals_do_not_construct_or_cache_probe_transports()
let error = request(&generation, "quic", &query).await.unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
assert!(local_failure(&error));
assert!(generation.probes.connectors.lock().is_empty(), "{url}");
assert!(
generation
.probes
.transports
.lock()
.connectors
.lock()
.is_empty(),
"{url}"
);
let error = request(&generation, "missing", &query).await.unwrap_err();
assert_eq!(
error.kind(),
@@ -2319,6 +2437,8 @@ async fn cached_connectors_reject_forbidden_hostname_candidates() {
for name in ["DIRECT", "http"] {
let cached = generation
.probes
.transports
.lock()
.connector(&generation.catalog, name)
.unwrap();
for _ in 0..2 {
@@ -2334,6 +2454,8 @@ async fn cached_connectors_reject_forbidden_hostname_candidates() {
&cached,
&generation
.probes
.transports
.lock()
.connector(&generation.catalog, name)
.unwrap()
));
@@ -2364,6 +2486,8 @@ async fn carrier_probe_checks_http_hostname_before_connecting() {
);
let cached = generation
.probes
.transports
.lock()
.connector(&generation.catalog, "outbound")
.unwrap();
for _ in 0..2 {
@@ -2382,6 +2506,8 @@ async fn carrier_probe_checks_http_hostname_before_connecting() {
&cached,
&generation
.probes
.transports
.lock()
.connector(&generation.catalog, "outbound")
.unwrap()
));
+2 -2
View File
@@ -140,7 +140,7 @@ impl Ingress {
.listeners
.iter()
.map(|listener| {
tokio::spawn(serve(
crate::network_tasks::spawn(serve(
listener.clone(),
selector.clone(),
resolver.clone(),
@@ -230,7 +230,7 @@ async fn serve(
};
let selector = selector.clone();
let resolver = resolver.clone();
connections.spawn(async move {
crate::network_tasks::spawn_in(&mut connections, async move {
let _permit = permit;
let _ = crate::tun::handle_tcp_connection(
stream,
+44 -15
View File
@@ -7,6 +7,7 @@ mod routes;
pub(crate) mod spec;
#[cfg(test)]
mod tests;
mod underlay;
use std::io;
use std::os::fd::BorrowedFd;
@@ -37,7 +38,9 @@ pub(crate) struct Session {
_connection: AbortOnDropHandle<()>,
_events: AbortOnDropHandle<()>,
changes: tokio::sync::mpsc::Receiver<()>,
underlay: Arc<underlay::Monitor>,
reconcile_after: tokio::time::Instant,
poll_at: tokio::time::Instant,
}
impl Session {
@@ -56,8 +59,15 @@ impl Session {
.map_err(io::Error::other)?;
let connection = AbortOnDropHandle::new(tokio::spawn(connection));
let (changed, changes) = tokio::sync::mpsc::channel(1);
let underlay = Arc::new(underlay::Monitor::default());
let monitor = underlay.clone();
let events = AbortOnDropHandle::new(tokio::spawn(async move {
while events.next().await.is_some() {
while let Some((message, _)) = events.next().await {
if let rtnetlink::packet_core::NetlinkPayload::InnerMessage(message) =
message.payload
{
monitor.event(&message);
}
let _ = changed.try_send(());
}
}));
@@ -119,7 +129,9 @@ impl Session {
_connection: connection,
_events: events,
changes,
underlay,
reconcile_after: tokio::time::Instant::now(),
poll_at: tokio::time::Instant::now() + Duration::from_secs(10),
})
}
@@ -170,6 +182,10 @@ impl Session {
self.ingress.clone()
}
pub(crate) fn spec(&self) -> &Spec {
&self.record.spec
}
fn firewall_policy(&self, active: bool) -> serde_json::Value {
nft::install(
&self.record.spec,
@@ -181,15 +197,6 @@ impl Session {
)
}
pub(crate) fn validate(&self, configs: &[crate::config::ResolvedConfig]) -> io::Result<()> {
if Spec::from_resolved_configs(configs)?.as_ref() != Some(&self.record.spec) {
return Err(spec::invalid(
"managed TUN topology changes require a restart",
));
}
Ok(())
}
pub(crate) async fn activate(&mut self) -> io::Result<()> {
routes::check(
&self.handle,
@@ -221,20 +228,39 @@ impl Session {
}
}
self.active = true;
self.underlay.initialize(
underlay::snapshot(
&self.handle,
self.record.ifindex.unwrap(),
self.record.spec.linux.bypass_mark,
)
.await?,
);
Ok(())
}
pub(crate) async fn changed(&mut self) {
tokio::time::sleep_until(self.reconcile_after).await;
tokio::select! {
_ = self.changes.recv() => {},
_ = tokio::time::sleep(Duration::from_secs(10)) => {},
_ = self.changes.recv() => {
self.poll_at = tokio::time::Instant::now() + Duration::from_millis(250);
tokio::time::sleep_until(self.poll_at).await;
},
_ = tokio::time::sleep_until(self.poll_at) => {},
}
}
pub(crate) async fn reconcile(&mut self) -> io::Result<()> {
pub(crate) fn available(&self) -> bool {
self.underlay.available()
}
pub(crate) fn refresh_bootstrap(&self) -> io::Result<()> {
self._policy.refresh_bootstrap()
}
pub(crate) async fn reconcile(&mut self) -> io::Result<bool> {
if !self.active {
return Ok(());
return Ok(false);
}
while self.changes.try_recv().is_ok() {}
let index = self.record.ifindex.unwrap();
@@ -259,7 +285,10 @@ impl Session {
self.on_link = networks;
}
self.reconcile_after = tokio::time::Instant::now() + Duration::from_secs(1);
Ok(())
self.poll_at = tokio::time::Instant::now() + Duration::from_secs(10);
Ok(self.underlay.update(
underlay::snapshot(&self.handle, index, self.record.spec.linux.bypass_mark).await?,
))
}
pub(crate) async fn deactivate(&mut self) -> io::Result<()> {
+16 -2
View File
@@ -71,7 +71,7 @@ pub(super) fn owned_link(link: &LinkMessage, index: u32, name: &str, alias: &str
.contains(&LinkAttribute::IfAlias(alias.to_owned()))
}
fn table(route: &RouteMessage) -> u32 {
pub(super) fn table(route: &RouteMessage) -> u32 {
route
.attributes
.iter()
@@ -82,7 +82,7 @@ fn table(route: &RouteMessage) -> u32 {
.unwrap_or(u32::from(route.header.table))
}
async fn dump(handle: &Handle, ipv6: bool) -> io::Result<Vec<RouteMessage>> {
pub(super) async fn dump(handle: &Handle, ipv6: bool) -> io::Result<Vec<RouteMessage>> {
let family = if ipv6 {
AddressFamily::Inet6
} else {
@@ -231,6 +231,20 @@ pub(super) async fn preflight(handle: &Handle, spec: &Spec) -> io::Result<()> {
Ok(())
}
pub(super) fn native_default(route: &RouteMessage) -> bool {
table(route) == 254
&& route.header.kind == RouteType::Unicast
&& route.header.destination_prefix_length == 0
&& route.header.source_prefix_length == 0
&& route.header.tos == 0
&& !route.attributes.iter().any(|attribute| {
matches!(
attribute,
RouteAttribute::Source(_) | RouteAttribute::Iif(_) | RouteAttribute::Mark(_)
)
})
}
pub(super) fn check_mark_reservation(rule: &RuleMessage, spec: &Spec) -> io::Result<()> {
let mark = rule
.attributes
+23 -4
View File
@@ -1,4 +1,5 @@
use super::*;
mod reconnect;
use serde_json::json;
mod fallback;
@@ -430,6 +431,17 @@ fn managed_kernel_capture_and_cleanup() -> io::Result<()> {
const PARENT: &str = "SHOES_MANAGED_PARENT_NETNS";
const MOUNT: &str = "SHOES_MANAGED_PARENT_MOUNTNS";
let namespace = std::fs::read_link("/proc/self/ns/net")?;
if std::env::var_os("SHOES_MANAGED_PEER_ECHO").is_some() {
assert_ne!(namespace.as_os_str(), std::env::var_os(PARENT).unwrap());
assert_ne!(
std::fs::read_link("/proc/self/ns/mnt")?.as_os_str(),
std::env::var_os(MOUNT).unwrap()
);
return tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?
.block_on(reconnect::echo());
}
if std::env::var_os("SHOES_ORPHAN_HELPER").is_some() {
assert_ne!(namespace.as_os_str(), std::env::var_os(PARENT).unwrap());
return tokio::runtime::Builder::new_current_thread().enable_all().build()?
@@ -586,7 +598,7 @@ fn managed_kernel_capture_and_cleanup() -> io::Result<()> {
}
let mut daemons = resolved_fixture().await?;
let result =
tokio::time::timeout(Duration::from_secs(90), kernel_capture()).await?;
tokio::time::timeout(Duration::from_secs(180), kernel_capture()).await?;
for child in &mut daemons {
let _ = child.kill().await;
let _ = child.wait().await;
@@ -620,7 +632,7 @@ fn managed_kernel_capture_and_cleanup() -> io::Result<()> {
"--",
"timeout",
"--kill-after=5s",
"120s",
"210s",
"unshare",
"--net",
"--mount",
@@ -1032,11 +1044,15 @@ async fn kernel_capture() -> io::Result<()> {
}
session.reconcile().await?;
let index = session.record.ifindex;
session.validate(&[crate::config::ResolvedConfig::TunServer(config.clone())])?;
let topology = crate::desktop::NetworkTopology {
spec: Some(session.spec().clone()),
ingress: session.ingress(),
};
topology.validate(&[crate::config::ResolvedConfig::TunServer(config.clone())])?;
let mut changed = config.clone();
changed.mtu += 1;
assert!(
session
topology
.validate(&[crate::config::ResolvedConfig::TunServer(changed)])
.is_err()
);
@@ -1089,6 +1105,9 @@ async fn kernel_capture() -> io::Result<()> {
"managed process recovery (redirect={redirect}): {error}"
))
})?;
reconnect::handover(redirect).await.map_err(|error| {
io::Error::other(format!("handover (redirect={redirect}): {error}"))
})?;
}
orphan_helper_recovery().await?;
preflight_failures().await?;
+386
View File
@@ -0,0 +1,386 @@
use super::*;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt};
pub(super) async fn echo() -> io::Result<()> {
let mut tasks = tokio::task::JoinSet::new();
for address in ["198.51.100.2:19123", "[fd00:2::2]:19123"] {
let tcp = tokio::net::TcpListener::bind(address).await?;
let udp = tokio::net::UdpSocket::bind(address).await?;
tasks.spawn(async move {
let mut flows = tokio::task::JoinSet::new();
loop {
let (mut stream, peer) = tokio::select! {
result = tcp.accept() => result.unwrap(),
_ = flows.join_next(), if !flows.is_empty() => continue,
};
flows.spawn(async move {
let mut packet = [0; 4];
while stream.read_exact(&mut packet).await.is_ok() {
if &packet != b"ping" {
break;
}
if stream
.write_all(format!("{}\n", peer.ip()).as_bytes())
.await
.is_err()
{
break;
}
}
});
}
});
tasks.spawn(async move {
let mut packet = [0; 4];
loop {
let (_, peer) = udp.recv_from(&mut packet).await.unwrap();
udp.send_to(peer.ip().to_string().as_bytes(), peer)
.await
.unwrap();
}
});
}
println!("Peer echo ready");
tasks.join_next().await;
Err(io::Error::other("echo service stopped"))
}
async fn line(
reader: &mut (impl tokio::io::AsyncBufRead + Unpin),
expected: &str,
) -> io::Result<()> {
tokio::time::timeout(Duration::from_secs(15), async {
let mut line = String::new();
loop {
line.clear();
if reader.read_line(&mut line).await? == 0 {
return Err(io::Error::other(format!("child exited before {expected}")));
}
if line.contains(expected) {
return Ok(());
}
}
})
.await?
}
async fn payload(sources: [&str; 2]) -> io::Result<()> {
for (target, expected) in ["198.51.100.2:19123", "[fd00:2::2]:19123"]
.into_iter()
.zip(sources)
{
let mut stream = tokio::net::TcpStream::connect(target).await?;
stream.write_all(b"ping").await?;
let mut reader = tokio::io::BufReader::new(stream);
let mut reply = String::new();
tokio::time::timeout(Duration::from_secs(3), reader.read_line(&mut reply)).await??;
assert_eq!(reply.trim(), expected);
let udp = tokio::net::UdpSocket::bind(if target.starts_with('[') {
"[::]:0"
} else {
"0.0.0.0:0"
})
.await?;
udp.send_to(b"ping", target).await?;
let mut bytes = [0; 128];
let length = tokio::time::timeout(Duration::from_secs(3), udp.recv(&mut bytes)).await??;
assert_eq!(std::str::from_utf8(&bytes[..length]).unwrap(), expected);
}
Ok(())
}
fn identity() -> io::Result<serde_json::Value> {
let links: serde_json::Value =
serde_json::from_str(&run("ip", &["-j", "link", "show", "shoes0"])?)?;
let tables: serde_json::Value = serde_json::from_str(&run("nft", &["-j", "list", "tables"])?)?;
Ok(json!({"index":links[0]["ifindex"], "alias":links[0]["ifalias"], "tables":tables}))
}
pub(super) async fn handover(redirect: bool) -> io::Result<()> {
run(
"ip",
&[
"link", "add", "native1", "type", "veth", "peer", "name", "remote1",
],
)?;
run("ip", &["link", "set", "remote1", "netns", "peer"])?;
for args in [
vec!["link", "set", "native1", "up"],
vec!["addr", "add", "192.0.3.1/24", "dev", "native1"],
vec![
"-6",
"addr",
"add",
"fd00:3::1/64",
"dev",
"native1",
"nodad",
],
vec!["-n", "peer", "link", "set", "remote1", "up"],
vec![
"-n",
"peer",
"addr",
"add",
"192.0.3.2/24",
"dev",
"remote1",
],
vec![
"-n",
"peer",
"-6",
"addr",
"add",
"fd00:3::2/64",
"dev",
"remote1",
"nodad",
],
] {
run("ip", &args)?;
}
let mut peer = tokio::process::Command::new("ip")
.args(["netns", "exec", "peer"])
.arg(std::env::current_exe()?)
.args(KERNEL_TEST_ARGS)
.env("SHOES_MANAGED_PEER_ECHO", "1")
.stdout(std::process::Stdio::piped())
.kill_on_drop(true)
.spawn()?;
line(
&mut tokio::io::BufReader::new(peer.stdout.take().unwrap()),
"Peer echo ready",
)
.await?;
let path = "/run/reconnect.yaml";
let unix_path = "/run/reconnect.sock";
let mut config = vec![
serde_json::to_value(configuration(redirect))?,
json!({"path":unix_path, "protocol":{"type":"socks"}}),
];
std::fs::write(
path,
serde_yaml::to_string(&config).map_err(io::Error::other)?,
)?;
let mut child = tokio::process::Command::new(std::env::current_exe()?)
.args(KERNEL_TEST_ARGS)
.env("SHOES_MANAGED_PROCESS_CONFIG", path)
.stdout(std::process::Stdio::piped())
.kill_on_drop(true)
.spawn()?;
let mut stdout = tokio::io::BufReader::new(child.stdout.take().unwrap());
line(&mut stdout, "Servers ready").await?;
let original = identity()?;
payload(["192.0.2.1", "fd00:1::1"]).await?;
let mut stale = tokio::net::TcpStream::connect("198.51.100.2:19123").await?;
stale.write_all(b"ping").await?;
let mut stale = tokio::io::BufReader::new(stale);
let mut reply = String::new();
stale.read_line(&mut reply).await?;
run("ip", &["link", "add", "unused0", "type", "dummy"])?;
run("ip", &["link", "set", "unused0", "up"])?;
run("ip", &["addr", "add", "192.0.99.1/24", "dev", "unused0"])?;
tokio::time::sleep(Duration::from_secs(2)).await;
stale.write_all(b"ping").await?;
reply.clear();
tokio::time::timeout(Duration::from_secs(3), stale.read_line(&mut reply)).await??;
assert_eq!(
reply.trim(),
"192.0.2.1",
"unrelated interface churn reset a live flow"
);
run("ip", &["link", "del", "unused0"])?;
std::fs::write(path, "invalid: [")?;
run("ip", &["route", "replace", "default", "via", "192.0.3.2"])?;
run(
"ip",
&["-6", "route", "replace", "default", "via", "fd00:3::2"],
)?;
line(&mut stdout, "Network reconnected").await?;
assert_eq!(identity()?, original);
let result = tokio::time::timeout(Duration::from_secs(3), stale.read(&mut [0])).await?;
assert!(matches!(result, Ok(0)) || result.is_err(), "{result:?}");
payload(["192.0.3.1", "fd00:3::1"]).await?;
// Both families disappear and return unchanged; recovery must not require a new config.
run("ip", &["route", "del", "default"])?;
run("ip", &["-6", "route", "del", "default"])?;
line(&mut stdout, "Network unavailable; waiting for underlay").await?;
assert!(child.try_wait()?.is_none());
assert_eq!(identity()?, original);
// A retryable bind failure must not remove capture or block a correcting reload.
std::fs::remove_file(unix_path)?;
std::fs::create_dir(unix_path)?;
run("ip", &["route", "add", "default", "via", "192.0.3.2"])?;
run("ip", &["-6", "route", "add", "default", "via", "fd00:3::2"])?;
assert!(
tokio::time::timeout(
Duration::from_secs(3),
line(&mut stdout, "Network reconnected")
)
.await
.is_err()
);
assert!(child.try_wait()?.is_none());
assert_eq!(identity()?, original);
config.pop();
config.push(json!({"address":"127.0.0.1:19124", "protocol":{"type":"socks"}}));
std::fs::write(
path,
serde_yaml::to_string(&config).map_err(io::Error::other)?,
)?;
assert_eq!(
unsafe { libc::kill(child.id().unwrap() as i32, libc::SIGHUP) },
0
);
line(&mut stdout, "Network reconnected").await?;
line(&mut stdout, "Servers ready").await?;
assert!(std::path::Path::new(unix_path).is_dir());
std::fs::remove_dir(unix_path)?;
let mut socks = tokio::net::TcpStream::connect("127.0.0.1:19124").await?;
socks.write_all(&[5, 1, 0]).await?;
let mut greeting = [0; 2];
tokio::time::timeout(Duration::from_secs(3), socks.read_exact(&mut greeting)).await??;
assert_eq!(greeting, [5, 0]);
drop(socks);
payload(["192.0.3.1", "fd00:3::1"]).await?;
// Same-interface source replacement within the existing subnet.
run("ip", &["addr", "add", "192.0.3.9/24", "dev", "native1"])?;
run(
"ip",
&[
"route",
"replace",
"default",
"via",
"192.0.3.2",
"src",
"192.0.3.9",
],
)?;
run(
"ip",
&[
"-6",
"addr",
"add",
"fd00:3::9/64",
"dev",
"native1",
"nodad",
],
)?;
run(
"ip",
&[
"-6",
"route",
"replace",
"default",
"via",
"fd00:3::2",
"src",
"fd00:3::9",
],
)?;
line(&mut stdout, "Network reconnected").await?;
payload(["192.0.3.9", "fd00:3::9"]).await?;
assert_eq!(identity()?, original);
// Multipath defaults use nested nexthops even with only one eligible path.
run(
"ip",
&[
"route",
"replace",
"default",
"src",
"192.0.3.9",
"nexthop",
"via",
"192.0.3.2",
"dev",
"native1",
],
)?;
run(
"ip",
&[
"-6",
"route",
"replace",
"default",
"src",
"fd00:3::9",
"nexthop",
"via",
"fd00:3::2",
"dev",
"native1",
],
)?;
run("ip", &["link", "set", "native1", "down"])?;
run("ip", &["link", "set", "native1", "up"])?;
// Link-down can remove IPv6 addresses; restore the same lease before its default.
for address in ["fd00:3::1/64", "fd00:3::9/64"] {
run(
"ip",
&["-6", "addr", "replace", address, "dev", "native1", "nodad"],
)?;
}
run(
"ip",
&[
"route",
"replace",
"default",
"src",
"192.0.3.9",
"nexthop",
"via",
"192.0.3.2",
"dev",
"native1",
],
)?;
run(
"ip",
&[
"-6",
"route",
"replace",
"default",
"src",
"fd00:3::9",
"nexthop",
"via",
"fd00:3::2",
"dev",
"native1",
],
)?;
line(&mut stdout, "Network reconnected").await?;
payload(["192.0.3.9", "fd00:3::9"]).await?;
run("ip", &["route", "del", "default"])?;
run("ip", &["-6", "route", "del", "default"])?;
line(&mut stdout, "Network unavailable; waiting for underlay").await?;
assert_eq!(
unsafe { libc::kill(child.id().unwrap() as i32, libc::SIGTERM) },
0
);
assert!(
tokio::time::timeout(Duration::from_secs(10), child.wait())
.await??
.success()
);
peer.kill().await?;
peer.wait().await?;
run("ip", &["link", "del", "native1"])?;
run("ip", &["route", "replace", "default", "via", "192.0.2.2"])?;
run(
"ip",
&["-6", "route", "replace", "default", "via", "fd00:1::2"],
)?;
Ok(())
}
+455
View File
@@ -0,0 +1,455 @@
use std::collections::BTreeSet;
use std::io;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use futures::TryStreamExt;
use parking_lot::Mutex;
use rtnetlink::packet_route::{
RouteNetlinkMessage,
address::{AddressAttribute, AddressFlags, AddressHeaderFlags, AddressMessage},
link::{LinkAttribute, LinkFlags},
route::{RouteAddress, RouteAttribute, RouteMessage, RouteVia},
};
use rtnetlink::{Handle, RouteMessageBuilder};
use super::routes;
#[derive(Clone, Debug, PartialEq, Eq)]
struct Path {
index: u32,
name: String,
gateway: Option<IpAddr>,
source: Option<IpAddr>,
preferred_source: Option<IpAddr>,
nexthop: Option<u32>,
metric: u32,
addresses: BTreeSet<IpAddr>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct Snapshot([Option<Path>; 2]);
impl Snapshot {
pub fn available(&self) -> bool {
self.0.iter().any(Option::is_some)
}
fn changed_from(&self, previous: &Self) -> bool {
self.0
.iter()
.zip(&previous.0)
.any(|(new, old)| match (new, old) {
(Some(new), Some(old)) => {
new.index != old.index
|| new.name != old.name
|| new.gateway != old.gateway
|| new.preferred_source != old.preferred_source
// A new privacy address can become preferred while the old source is valid.
|| old
.source
.is_some_and(|source| !new.addresses.contains(&source))
}
(None, None) => false,
_ => true,
})
}
}
#[derive(Default)]
struct State {
snapshot: Snapshot,
disrupted: bool,
}
#[derive(Default)]
pub(super) struct Monitor(Mutex<State>);
impl Monitor {
pub fn initialize(&self, snapshot: Snapshot) {
*self.0.lock() = State {
snapshot,
disrupted: false,
};
}
pub fn update(&self, snapshot: Snapshot) -> bool {
let mut state = self.0.lock();
let changed = snapshot.changed_from(&state.snapshot) || state.disrupted;
state.snapshot = snapshot;
state.disrupted = false;
changed
}
pub fn available(&self) -> bool {
self.0.lock().snapshot.available()
}
// Retains losses even when a down/up burst finishes before the next dump.
pub fn event(&self, message: &RouteNetlinkMessage) {
let mut state = self.0.lock();
let lost = state.snapshot.0.iter().flatten().any(|path| match message {
RouteNetlinkMessage::DelLink(link) => link.header.index == path.index,
RouteNetlinkMessage::NewLink(link) => {
link.header.index == path.index
&& !link
.header
.flags
.contains(LinkFlags::Up | LinkFlags::LowerUp)
}
RouteNetlinkMessage::DelAddress(address) => {
address.header.index == path.index && address.attributes.iter().any(|attribute| {
matches!(attribute, AddressAttribute::Local(ip) | AddressAttribute::Address(ip)
if Some(*ip) == path.source)
})
}
RouteNetlinkMessage::DelRoute(route) => {
routes::native_default(route)
&& matches_path(route, path.index, path.gateway, path.nexthop)
&& metric(route) == path.metric
}
_ => false,
});
state.disrupted |= lost;
}
}
fn metric(route: &RouteMessage) -> u32 {
route
.attributes
.iter()
.find_map(|attribute| match attribute {
RouteAttribute::Priority(priority) => Some(*priority),
_ => None,
})
.unwrap_or(0)
}
fn route_ip(route: &RouteMessage, source: bool) -> Option<IpAddr> {
attribute_ip(&route.attributes, source)
}
fn attribute_ip(attributes: &[RouteAttribute], source: bool) -> Option<IpAddr> {
attributes.iter().find_map(|attribute| {
let address = match attribute {
RouteAttribute::PrefSource(address) if source => address,
RouteAttribute::Gateway(address) if !source => address,
RouteAttribute::Via(RouteVia::Inet(ip)) if !source => return Some((*ip).into()),
RouteAttribute::Via(RouteVia::Inet6(ip)) if !source => return Some((*ip).into()),
_ => return None,
};
match address {
RouteAddress::Inet(ip) => Some(IpAddr::V4(*ip)),
RouteAddress::Inet6(ip) => Some(IpAddr::V6(*ip)),
_ => None,
}
})
}
fn nexthop(route: &RouteMessage) -> Option<u32> {
route
.attributes
.iter()
.find_map(|attribute| match attribute {
RouteAttribute::NhId(id) => Some(*id),
_ => None,
})
}
fn matches_path(
route: &RouteMessage,
index: u32,
gateway: Option<IpAddr>,
nh: Option<u32>,
) -> bool {
(route.attributes.contains(&RouteAttribute::Oif(index)) && route_ip(route, false) == gateway)
|| nh.is_some_and(|id| nexthop(route) == Some(id))
|| route.attributes.iter().any(|attribute| match attribute {
RouteAttribute::MultiPath(hops) => hops.iter().any(|hop| {
hop.interface_index == index && attribute_ip(&hop.attributes, false) == gateway
}),
_ => false,
})
}
fn usable_addresses(addresses: &[AddressMessage], index: u32, ipv6: bool) -> BTreeSet<IpAddr> {
addresses.iter().filter(|address| {
address.header.index == index
&& !address.header.flags.intersects(AddressHeaderFlags::Tentative | AddressHeaderFlags::Dadfailed)
&& !address.attributes.iter().any(|attribute| matches!(attribute,
AddressAttribute::Flags(flags) if flags.intersects(AddressFlags::Tentative | AddressFlags::Dadfailed)))
}).filter_map(|address| {
address.attributes.iter().find_map(|attribute| match attribute {
AddressAttribute::Local(ip) => Some(*ip),
_ => None,
}).or_else(|| address.attributes.iter().find_map(|attribute| match attribute {
AddressAttribute::Address(ip) => Some(*ip),
_ => None,
}))
}).filter(|ip| ip.is_ipv6() == ipv6 && !ip.is_loopback() && !ip.is_unspecified()).collect()
}
pub(super) async fn snapshot(handle: &Handle, owned_index: u32, mark: u32) -> io::Result<Snapshot> {
let links = routes::request(handle.link().get().execute().try_collect::<Vec<_>>()).await?;
let addresses =
routes::request(handle.address().get().execute().try_collect::<Vec<_>>()).await?;
let mut snapshot = Snapshot::default();
for ipv6 in [false, true] {
let defaults = routes::dump(handle, ipv6).await?;
if !defaults.iter().any(routes::native_default) {
continue;
}
// Read-only lookups select the kernel's actual next hop and source. No probe is sent.
let query = if ipv6 {
RouteMessageBuilder::<Ipv6Addr>::new()
.destination_prefix("2606:4700:4700::1111".parse().unwrap(), 128)
.mark(mark)
.build()
} else {
RouteMessageBuilder::<Ipv4Addr>::new()
.destination_prefix(Ipv4Addr::new(1, 1, 1, 1), 32)
.mark(mark)
.build()
};
let mut response = handle.route().get(query).execute();
let lookup = response.try_next();
let route = match tokio::time::timeout(std::time::Duration::from_secs(10), lookup).await {
Ok(Ok(Some(route))) => route,
Ok(Ok(None)) => continue,
Ok(Err(rtnetlink::Error::NetlinkError(_))) => continue,
Ok(Err(error)) => return Err(io::Error::other(error)),
Err(_) => {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"underlay lookup timed out",
));
}
};
let Some(index) = route
.attributes
.iter()
.find_map(|attribute| match attribute {
RouteAttribute::Oif(index) => Some(*index),
_ => None,
})
else {
continue;
};
if index == owned_index || routes::table(&route) != 254 {
continue;
}
let gateway = route_ip(&route, false);
let nexthop = nexthop(&route);
let Some(default) = defaults
.iter()
.filter(|route| {
routes::native_default(route) && matches_path(route, index, gateway, nexthop)
})
.min_by_key(|route| metric(route))
else {
continue;
};
let Some(link) = links.iter().find(|link| link.header.index == index) else {
continue;
};
if !link
.header
.flags
.contains(LinkFlags::Up | LinkFlags::LowerUp)
{
continue;
}
let Some(name) = link
.attributes
.iter()
.find_map(|attribute| match attribute {
LinkAttribute::IfName(name) => Some(name.clone()),
_ => None,
})
else {
continue;
};
let addresses = usable_addresses(&addresses, index, ipv6);
if addresses.is_empty() {
continue;
}
snapshot.0[usize::from(ipv6)] = Some(Path {
index,
name,
gateway,
source: route_ip(&route, true),
preferred_source: route_ip(default, true),
nexthop,
metric: metric(default),
addresses,
});
}
Ok(snapshot)
}
#[cfg(test)]
mod tests {
use super::*;
fn sample(ipv6: bool) -> Snapshot {
let mut snapshot = Snapshot::default();
let source: IpAddr = if ipv6 { "2001:db8::1" } else { "192.0.2.1" }
.parse()
.unwrap();
snapshot.0[usize::from(ipv6)] = Some(Path {
index: 2,
name: "native0".into(),
gateway: Some(
if ipv6 { "2001:db8::2" } else { "192.0.2.2" }
.parse()
.unwrap(),
),
source: Some(source),
preferred_source: None,
nexthop: None,
metric: 100,
addresses: [source].into(),
});
snapshot
}
#[test]
fn identity_source_gateway_and_availability_changes_reset() {
for ipv6 in [false, true] {
let initial = sample(ipv6);
let monitor = Monitor::default();
monitor.initialize(initial.clone());
assert!(!monitor.update(initial.clone()));
for mutation in 0..4 {
let mut changed = initial.clone();
let path = changed.0[usize::from(ipv6)].as_mut().unwrap();
match mutation {
0 => path.index += 1,
1 => path.name = "native1".into(),
2 => {
path.gateway = Some(
if ipv6 { "2001:db8::3" } else { "192.0.2.3" }
.parse()
.unwrap(),
)
}
_ => {
path.source = Some(
if ipv6 { "2001:db8::4" } else { "192.0.2.4" }
.parse()
.unwrap(),
);
path.addresses = [path.source.unwrap()].into();
}
}
monitor.initialize(initial.clone());
assert!(monitor.update(changed.clone()));
assert!(!monitor.update(changed));
}
monitor.initialize(initial.clone());
assert!(monitor.update(Snapshot::default()));
assert!(!monitor.available());
assert!(!monitor.update(Snapshot::default()));
assert!(monitor.update(initial));
assert!(monitor.available());
}
}
#[test]
fn metadata_and_unused_addresses_do_not_reset_but_source_removal_does() {
let monitor = Monitor::default();
let initial = sample(false);
monitor.initialize(initial.clone());
let mut changed = initial;
let path = changed.0[0].as_mut().unwrap();
path.metric = 200;
path.addresses.insert("192.0.2.4".parse().unwrap());
assert!(!monitor.update(changed.clone()));
changed.0[0]
.as_mut()
.unwrap()
.addresses
.remove(&"192.0.2.1".parse().unwrap());
assert!(monitor.update(changed));
}
#[test]
fn privacy_rotation_keeps_old_flows_but_explicit_source_changes_reset() {
let monitor = Monitor::default();
let initial = sample(true);
monitor.initialize(initial.clone());
let mut rotated = initial.clone();
let path = rotated.0[1].as_mut().unwrap();
path.source = Some("2001:db8::9".parse().unwrap());
path.addresses.insert(path.source.unwrap());
assert!(!monitor.update(rotated.clone()));
rotated.0[1]
.as_mut()
.unwrap()
.addresses
.remove(&"2001:db8::1".parse().unwrap());
assert!(!monitor.update(rotated.clone()));
rotated.0[1].as_mut().unwrap().preferred_source = Some("2001:db8::9".parse().unwrap());
assert!(monitor.update(rotated));
}
#[test]
fn multipath_defaults_match_the_selected_hop_and_preserve_loss_events() {
use rtnetlink::packet_route::route::{RouteNextHop, RouteType};
let initial = sample(false);
let path = initial.0[0].as_ref().unwrap();
let mut route = RouteMessage::default();
route.header.table = 254;
route.header.kind = RouteType::Unicast;
let mut hop = RouteNextHop::default();
hop.interface_index = path.index;
hop.attributes
.push(RouteAttribute::Gateway(RouteAddress::Inet(
"192.0.2.2".parse().unwrap(),
)));
route.attributes = vec![
RouteAttribute::MultiPath(vec![hop]),
RouteAttribute::Priority(100),
];
assert!(matches_path(&route, path.index, path.gateway, None));
assert!(!matches_path(&route, 99, path.gateway, None));
let monitor = Monitor::default();
monitor.initialize(initial.clone());
monitor.event(&RouteNetlinkMessage::DelRoute(route));
assert!(monitor.update(initial));
}
#[test]
fn short_outages_survive_event_coalescing_without_tracking_unrelated_links() {
let monitor = Monitor::default();
let initial = sample(false);
monitor.initialize(initial.clone());
let mut link = rtnetlink::packet_route::link::LinkMessage::default();
link.header.index = 3;
monitor.event(&RouteNetlinkMessage::DelLink(link.clone()));
assert!(!monitor.update(initial.clone()));
link.header.index = 2;
monitor.event(&RouteNetlinkMessage::NewLink(link.clone()));
link.header.flags = LinkFlags::Up | LinkFlags::LowerUp;
monitor.event(&RouteNetlinkMessage::NewLink(link));
assert!(monitor.update(initial.clone()));
assert!(!monitor.update(initial));
}
#[test]
fn unusable_addresses_and_lifetime_countdowns_are_ignored() {
let mut address = AddressMessage::default();
address.header.index = 2;
address
.attributes
.push(AddressAttribute::Address("2001:db8::1".parse().unwrap()));
assert_eq!(usable_addresses(&[address.clone()], 2, true).len(), 1);
assert!(usable_addresses(&[address.clone()], 3, true).is_empty());
address.header.flags = AddressHeaderFlags::Tentative;
assert!(usable_addresses(&[address.clone()], 2, true).is_empty());
address.header.flags = AddressHeaderFlags::empty();
address
.attributes
.push(AddressAttribute::Flags(AddressFlags::Dadfailed));
assert!(usable_addresses(&[address], 2, true).is_empty());
}
}
+97 -22
View File
@@ -6,19 +6,70 @@ pub(crate) const MANAGED_STATE_DIRECTORY: &str = "/run/shoes";
use std::io;
#[cfg(test)]
tokio::task_local! {
pub(crate) static TEST_NETWORK: std::cell::RefCell<Option<NetworkSession>>;
}
pub(crate) struct NetworkTopology {
#[cfg(target_os = "linux")]
spec: Option<linux::Spec>,
#[cfg(target_os = "linux")]
ingress: Option<std::sync::Arc<linux::Ingress>>,
}
impl NetworkTopology {
pub(crate) fn validate(&self, configs: &[crate::config::ResolvedConfig]) -> io::Result<()> {
#[cfg(target_os = "linux")]
if linux::Spec::from_resolved_configs(configs)? != self.spec {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
if self.spec.is_none() {
"enabling managed TUN requires a restart"
} else {
"managed TUN topology changes require a restart"
},
));
}
let _ = configs;
Ok(())
}
#[cfg(target_os = "linux")]
pub(crate) fn ingress(&self) -> Option<std::sync::Arc<linux::Ingress>> {
self.ingress.clone()
}
}
#[derive(Default)]
pub(crate) struct NetworkSession {
#[cfg(test)]
test_changed: bool,
#[cfg(test)]
test_available: Option<bool>,
#[cfg(target_os = "linux")]
session: Option<linux::Session>,
#[cfg(test)]
test_events: Option<(
tokio::sync::mpsc::Receiver<()>,
tokio::sync::mpsc::Receiver<Option<bool>>,
tokio::sync::mpsc::Sender<()>,
)>,
}
impl NetworkSession {
pub(crate) fn topology(&self) -> NetworkTopology {
NetworkTopology {
#[cfg(target_os = "linux")]
spec: self.session.as_ref().map(|session| session.spec().clone()),
#[cfg(target_os = "linux")]
ingress: self.ingress(),
}
}
pub(crate) async fn prepare(configs: &[crate::config::Config]) -> io::Result<Self> {
#[cfg(test)]
if let Ok(Some(network)) = TEST_NETWORK.try_with(|network| network.borrow_mut().take()) {
return Ok(network);
}
#[cfg(target_os = "linux")]
{
if let Some(spec) = linux::Spec::from_configs(configs)? {
@@ -26,6 +77,10 @@ impl NetworkSession {
session: Some(linux::Session::prepare(spec).await?),
#[cfg(test)]
test_events: None,
#[cfg(test)]
test_changed: false,
#[cfg(test)]
test_available: None,
});
}
}
@@ -33,20 +88,16 @@ impl NetworkSession {
Ok(Self::default())
}
pub(crate) fn validate(&self, configs: &[crate::config::ResolvedConfig]) -> io::Result<()> {
#[cfg(target_os = "linux")]
match &self.session {
Some(session) => return session.validate(configs),
None if linux::Spec::from_resolved_configs(configs)?.is_some() => {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"enabling managed TUN requires a restart",
));
}
_ => {}
pub(crate) fn monitors_underlay(&self) -> bool {
#[cfg(test)]
if self.test_events.is_some() {
return true;
}
let _ = configs;
Ok(())
#[cfg(target_os = "linux")]
if self.session.is_some() {
return true;
}
false
}
#[cfg(target_os = "linux")]
@@ -73,7 +124,11 @@ impl NetworkSession {
pub(crate) async fn changed(&mut self) {
#[cfg(test)]
if let Some((changes, _)) = &mut self.test_events {
changes.recv().await.unwrap();
let event = changes.recv().await.unwrap();
self.test_changed = event.is_some();
if let Some(available) = event {
self.test_available = Some(available);
}
return;
}
#[cfg(target_os = "linux")]
@@ -83,22 +138,42 @@ impl NetworkSession {
std::future::pending().await
}
pub(crate) async fn reconcile(&mut self) -> io::Result<()> {
pub(crate) fn available(&self) -> bool {
#[cfg(test)]
if let Some((_, reconciled)) = &self.test_events {
reconciled.send(()).await.unwrap();
return Ok(());
if let Some(available) = self.test_available {
return available;
}
#[cfg(target_os = "linux")]
if let Some(session) = &mut self.session {
session.reconcile().await?;
if let Some(session) = &self.session {
return session.available();
}
true
}
pub(crate) fn refresh_bootstrap(&self) -> io::Result<()> {
#[cfg(target_os = "linux")]
if let Some(session) = &self.session {
session.refresh_bootstrap()?;
}
Ok(())
}
pub(crate) async fn reconcile(&mut self) -> io::Result<bool> {
#[cfg(test)]
if let Some((_, reconciled)) = &self.test_events {
let _ = reconciled.try_send(());
return Ok(std::mem::take(&mut self.test_changed));
}
#[cfg(target_os = "linux")]
if let Some(session) = &mut self.session {
return session.reconcile().await;
}
Ok(false)
}
#[cfg(test)]
pub(crate) fn with_test_events(
changes: tokio::sync::mpsc::Receiver<()>,
changes: tokio::sync::mpsc::Receiver<Option<bool>>,
reconciled: tokio::sync::mpsc::Sender<()>,
) -> Self {
Self {
+2 -2
View File
@@ -46,7 +46,7 @@ pub(crate) fn controlled(endpoints: &[SocketAddr]) -> io::Result<Arc<dyn Resolve
NameServerConfig::new(endpoint.ip(), true, connections.to_vec())
})
.collect();
Ok(Arc::new(
Ok(super::network::scoped(Arc::new(
super::hickory_resolver::HickoryResolver::build_with_ns_configs(
servers,
chain,
@@ -54,7 +54,7 @@ pub(crate) fn controlled(endpoints: &[SocketAddr]) -> io::Result<Arc<dyn Resolve
Default::default(),
"managed bootstrap".to_owned(),
)?,
))
)))
}
#[cfg(test)]
+12 -3
View File
@@ -96,6 +96,15 @@ pub struct DnsRegistry {
}
impl DnsRegistry {
pub(crate) fn retained_state(&self) -> Self {
Self {
limits: self.limits,
forbidden: self.forbidden.clone(),
pools: self.pools.clone(),
..Self::new()
}
}
#[cfg(unix)]
pub(crate) fn for_rule_bootstrap(
configs: &[crate::config::Config],
@@ -195,7 +204,7 @@ impl DnsRegistry {
/// Register a DNS group.
pub fn register(&mut self, name: String, resolver: Arc<dyn Resolver>) {
self.groups.insert(name, resolver);
self.groups.insert(name, super::network::scoped(resolver));
}
/// Get a resolver by group name, returns None if not found.
@@ -213,14 +222,14 @@ impl DnsRegistry {
let forbidden = self.forbidden.clone();
self.default_resolver
.get_or_insert_with(|| {
real_only(
super::network::scoped(real_only(
Arc::new(SystemResolver::new(
Arc::new(CachingNativeResolver::new()),
forbidden.clone(),
Duration::from_secs(5),
)),
forbidden,
)
))
})
.clone()
}
+46
View File
@@ -553,6 +553,52 @@ async fn fake_context_reload_sharing_disable_and_reenable_keep_pool_identity() {
assert_eq!(context.restore(ip).unwrap().as_deref(), Some("stable.test"));
}
#[tokio::test]
async fn network_rebuild_retires_resolvers_but_keeps_live_fake_ip_pools() {
let old = crate::network_tasks::NetworkLifetime::default();
let registry = old
.scope(build_dns_registry(fake_groups(true)))
.await
.unwrap();
let resolver = registry.get_by_name("fake").unwrap();
let context = resolver.fake_dns_context().unwrap();
let ip = context.allocate("network.test", false).unwrap();
let retained = registry.retained_state();
old.cancel();
old.wait().await;
assert_eq!(
resolver
.resolve_location(
&crate::address::NetLocation::from_str("network.test:80", None).unwrap()
)
.await
.unwrap_err()
.kind(),
std::io::ErrorKind::Interrupted
);
let fresh = crate::network_tasks::NetworkLifetime::default();
let rebuilt = fresh
.scope(build_dns_registry_reusing(
fake_groups(true),
registry.limits,
Some(&retained),
))
.await
.unwrap();
assert!(Arc::ptr_eq(
&registry.pools["fake"].pool,
&rebuilt.pools["fake"].pool
));
let next = rebuilt
.get_by_name("fake")
.unwrap()
.fake_dns_context()
.unwrap();
assert_eq!(next.restore(ip).unwrap().as_deref(), Some("network.test"));
assert_eq!(next.allocate("network.test", false).unwrap(), ip);
assert!(next.allocate("after.test", false).is_ok());
}
#[tokio::test]
async fn equivalent_pool_ranges_reuse_live_mappings_on_reload() {
let mut groups = fake_groups(true);
+1 -1
View File
@@ -323,7 +323,7 @@ fn connection_setup(
deadline: tokio::time::Instant,
) -> Connection {
// A losing query must not cancel a shared handshake. Dropping its cache owner still does.
let task = tokio_util::task::AbortOnDropHandle::new(tokio::spawn(async move {
let task = tokio_util::task::AbortOnDropHandle::new(crate::network_tasks::spawn(async move {
tokio::time::timeout_at(deadline, future)
.await
.map_err(|_| {
+1
View File
@@ -20,6 +20,7 @@ pub mod fake_ip;
mod guard;
mod hickory_resolver;
mod message;
mod network;
mod parsed;
mod proxy_runtime;
mod stream;
+62
View File
@@ -0,0 +1,62 @@
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::sync::Arc;
use crate::network_tasks::NetworkTasks;
use crate::resolver::{DnsExchangeFuture, ResolveFuture, Resolver};
#[derive(Debug)]
struct NetworkResolver {
inner: Arc<dyn Resolver>,
network: NetworkTasks,
}
pub(super) fn scoped(inner: Arc<dyn Resolver>) -> Arc<dyn Resolver> {
match NetworkTasks::current() {
Some(network) => Arc::new(NetworkResolver { inner, network }),
None => inner,
}
}
impl NetworkResolver {
fn run<T: Send + 'static>(
&self,
operation: impl FnOnce(Arc<dyn Resolver>) -> Pin<Box<dyn Future<Output = io::Result<T>> + Send>>
+ Send
+ 'static,
) -> Pin<Box<dyn Future<Output = io::Result<T>> + Send>> {
let inner = self.inner.clone();
let network = self.network.clone();
Box::pin(async move {
tokio::select! {
biased;
_ = network.cancelled.cancelled() => Err(io::Error::new(io::ErrorKind::Interrupted, "DNS network retired")),
result = network.scope(async move { operation(inner).await }) => result,
}
})
}
}
impl Resolver for NetworkResolver {
fn resolve_location(&self, location: &crate::address::NetLocation) -> ResolveFuture {
let location = location.clone();
self.run(move |inner| inner.resolve_location(&location))
}
fn exchange(&self, request: Vec<u8>) -> DnsExchangeFuture {
self.run(move |inner| inner.exchange(request))
}
fn exchange_with_deadline(
&self,
request: Vec<u8>,
deadline: tokio::time::Instant,
) -> DnsExchangeFuture {
self.run(move |inner| inner.exchange_with_deadline(request, deadline))
}
fn fake_dns_context(&self) -> Option<Arc<super::DnsContext>> {
self.inner.fake_dns_context()
}
}
+12 -3
View File
@@ -30,6 +30,7 @@ pub struct ProxyRuntimeProvider {
quic_binder: ProxyQuicBinder,
/// Timeout for establishing connections to DNS upstreams.
connect_timeout: Duration,
network: Option<crate::network_tasks::NetworkTasks>,
}
impl ProxyRuntimeProvider {
@@ -49,6 +50,7 @@ impl ProxyRuntimeProvider {
bootstrap_resolver,
quic_binder,
connect_timeout,
network: crate::network_tasks::NetworkTasks::current(),
}
}
@@ -65,11 +67,18 @@ impl ProxyRuntimeProvider {
/// Spawn handle for tokio runtime.
#[derive(Clone, Default)]
pub struct TokioSpawnHandle;
pub struct TokioSpawnHandle(Option<crate::network_tasks::NetworkTasks>);
impl Spawn for TokioSpawnHandle {
fn spawn_bg(&mut self, future: impl Future<Output = ()> + Send + 'static) {
tokio::spawn(future);
match &self.0 {
Some(network) => {
network.spawn(future);
}
None => {
crate::network_tasks::spawn(future);
}
}
}
}
@@ -83,7 +92,7 @@ impl RuntimeProvider for ProxyRuntimeProvider {
type Tcp = ProxiedTcp;
fn create_handle(&self) -> Self::Handle {
TokioSpawnHandle
TokioSpawnHandle(self.network.clone())
}
fn connect_tcp(
+1 -1
View File
@@ -122,7 +122,7 @@ fn test_provider_is_clone() {
#[test]
fn test_spawn_handle_is_clone() {
let handle = TokioSpawnHandle;
let handle = TokioSpawnHandle::default();
let _cloned = handle.clone();
}
+1 -1
View File
@@ -162,7 +162,7 @@ pub(super) struct TcpDnsStream {
impl TcpDnsStream {
pub fn new(context: Arc<DnsContext>) -> Self {
let (stream, mut peer) = tokio::io::duplex(16 * 1024);
let task = tokio::spawn(async move {
let task = crate::network_tasks::spawn(async move {
let _ = serve_tcp(&mut peer, context).await;
});
Self { stream, task }
+2 -2
View File
@@ -125,7 +125,7 @@ impl H2MuxClientSession {
let (drain_tx, drain_rx) = oneshot::channel();
// Spawn connection driver
let abort_handle = tokio::spawn(async move {
let abort_handle = crate::network_tasks::spawn(async move {
// Dropping the last owner must still flush queued DATA and END_STREAM.
let result = tokio::select! {
result = &mut connection => Some(result),
@@ -167,7 +167,7 @@ impl H2MuxClientSession {
is_closed: Arc<AtomicBool>,
driver: tokio::task::AbortHandle,
) -> tokio::task::AbortHandle {
tokio::spawn(async move {
crate::network_tasks::spawn(async move {
let mut timer = interval(PING_INTERVAL);
timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
// Skip the first tick which returns immediately
+5 -5
View File
@@ -145,7 +145,7 @@ impl H2MuxServerSession {
// Spawn acceptor task with idle timeout monitoring
let is_closed_clone = Arc::clone(&is_closed);
let driver = tokio::spawn(async move {
let driver = crate::network_tasks::spawn(async move {
Self::accept_loop(connection, inbound_tx, is_closed_clone, activity).await;
});
@@ -188,7 +188,7 @@ impl H2MuxServerSession {
Some(Ok((request, respond))) => {
let inbound_tx = inbound_tx.clone();
let slots = slots.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
if let Err(e) = Self::handle_stream(request, respond, inbound_tx, slots).await {
debug!("H2MuxServerSession: stream error: {}", e);
}
@@ -227,7 +227,7 @@ impl H2MuxServerSession {
Ok((request, respond)) => {
let inbound_tx = inbound_tx.clone();
let slots = slots.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
if let Err(e) = Self::handle_stream(request, respond, inbound_tx, slots).await {
debug!("H2MuxServerSession: stream error during drain: {}", e);
}
@@ -367,7 +367,7 @@ where
};
let Some(inbound) = inbound else { break };
let fut = handler(inbound);
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
if let Err(e) = fut.await {
debug!("H2MUX stream handler error: {}", e);
}
@@ -415,7 +415,7 @@ where
let proxy_selector = proxy_selector.clone();
let resolver = resolver.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
if let Err(e) =
handle_h2mux_stream(inbound, udp_enabled, proxy_selector, resolver).await
{
+1 -1
View File
@@ -100,7 +100,7 @@ impl H2ClientSession {
let driver_retired = retired.clone();
let active = Arc::new(ActiveStreams::default());
let driver_active = active.clone();
let abort_handle = tokio::spawn(async move {
let abort_handle = crate::network_tasks::spawn(async move {
let completed = tokio::select! {
result = &mut connection => Some(result),
_ = &mut drain_rx => None,
+1 -1
View File
@@ -81,7 +81,7 @@ pub(crate) async fn serve(
};
let selector = selector.clone();
let resolver = resolver.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
let _permits = (local, global);
if let Err(error) = tunnel(request, respond, target, selector, resolver).await {
log::debug!("HTTP/2 CONNECT stream ended: {error}");
+3 -3
View File
@@ -272,7 +272,7 @@ impl UdpSession {
let send_socket = client_socket.clone();
let cancel_token = session_cancel_token.clone();
let task = tokio::spawn(async move {
let task = crate::network_tasks::spawn(async move {
if let Err(e) = run_udp_remote_to_local_loop(
session_id,
connection,
@@ -541,7 +541,7 @@ async fn run_tcp_loop(
let client_proxy_selector = client_proxy_selector.clone();
let resolver = resolver.clone();
let peer = connection.remote_address();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
let _permit = permit;
if let Err(e) = process_tcp_stream(
client_proxy_selector,
@@ -865,7 +865,7 @@ pub async fn start_hysteria2_server(
let resolver = resolver.clone();
let client_proxy_selector = client_proxy_selector.clone();
let hysteria2_password = hysteria2_password.clone();
let join_handle = tokio::spawn(async move {
let join_handle = crate::network_tasks::spawn(async move {
let mut tasks = crate::listener_tasks::ListenerTasks::immediate();
loop {
let conn = tokio::select! {
+1
View File
@@ -60,6 +60,7 @@ mod hysteria2_server;
mod listener_tasks;
mod mixed_handler;
mod naiveproxy;
mod network_tasks;
mod ntp;
mod option_util;
mod port_forward_handler;
+4 -4
View File
@@ -118,7 +118,7 @@ impl ListenerTasks {
}
pub fn spawn(&mut self, future: impl Future<Output = ()> + Send + 'static) {
self.tasks.spawn(future);
crate::network_tasks::spawn_in(&mut self.tasks, future);
}
pub fn is_empty(&self) -> bool {
@@ -137,12 +137,12 @@ impl Drop for ListenerTasks {
if self.tasks.is_empty() || self.grace.is_zero() {
return;
}
let Ok(runtime) = tokio::runtime::Handle::try_current() else {
if tokio::runtime::Handle::try_current().is_err() {
return;
};
}
let mut tasks = std::mem::take(&mut self.tasks);
let deadline = tokio::time::Instant::now() + self.grace;
runtime.spawn(async move {
crate::network_tasks::spawn(async move {
let drained = tokio::time::timeout_at(deadline, async {
while let Some(result) = tasks.join_next().await {
if let Err(error) = result {
+1 -1
View File
@@ -72,7 +72,7 @@ impl NaiveClientSession {
errors.start_driver();
let (drain_tx, drain_rx) = oneshot::channel();
let abort_handle = tokio::spawn(async move {
let abort_handle = crate::network_tasks::spawn(async move {
// Retired sessions must flush queued DATA and END_STREAM before teardown.
let result = tokio::select! {
result = &mut connection => Some(result),
+27 -1
View File
@@ -67,6 +67,32 @@ impl tokio::io::AsyncRead for HyperUpgradedStream {
mod fallback_tests {
use super::*;
#[tokio::test]
async fn executor_propagates_network_retirement_to_descendants() {
let network = crate::network_tasks::NetworkLifetime::default();
let tasks = Arc::new(Mutex::new(JoinSet::new()));
let executor = ConnectionExecutor(Arc::downgrade(&tasks));
let marker = Arc::new(());
let retained = marker.clone();
let (ready, started) = tokio::sync::oneshot::channel();
network
.scope(async {
hyper::rt::Executor::execute(&executor, async move {
assert!(crate::network_tasks::NetworkTasks::current().is_some());
crate::network_tasks::spawn(async move {
let _retained = retained;
ready.send(()).unwrap();
std::future::pending::<()>().await;
});
});
})
.await;
started.await.unwrap();
network.cancel();
network.wait().await;
assert_eq!(Arc::strong_count(&marker), 1);
}
#[tokio::test]
async fn connection_executor_cancels_children_without_a_reference_cycle() {
let tasks = Arc::new(Mutex::new(JoinSet::new()));
@@ -185,7 +211,7 @@ where
if let Some(tasks) = self.0.upgrade() {
let mut tasks = tasks.lock();
while tasks.try_join_next().is_some() {}
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
let _ = future.await;
});
}
+268
View File
@@ -0,0 +1,268 @@
use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, Weak};
use parking_lot::Mutex;
use tokio::task::{AbortHandle, JoinHandle, JoinSet};
use tokio_util::task::TaskTracker;
tokio::task_local! {
static CURRENT: NetworkTasks;
}
#[derive(Debug, Default)]
struct State {
cancelled: bool,
next_id: u64,
tasks: HashMap<u64, Option<AbortHandle>>,
}
/// One underlay lifetime, shared across ordinary configuration reloads.
#[derive(Clone, Debug, Default)]
pub(crate) struct NetworkTasks {
untracked: bool,
state: Arc<Mutex<State>>,
tracker: TaskTracker,
pub(crate) cancelled: tokio_util::sync::CancellationToken,
}
#[derive(Default)]
pub(crate) struct NetworkLifetime(pub NetworkTasks);
impl NetworkLifetime {
pub fn untracked() -> Self {
Self(NetworkTasks {
untracked: true,
..NetworkTasks::default()
})
}
}
impl std::ops::Deref for NetworkLifetime {
type Target = NetworkTasks;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl Drop for NetworkLifetime {
fn drop(&mut self) {
self.0.cancel();
}
}
struct Registration {
state: Weak<Mutex<State>>,
id: u64,
}
impl Drop for Registration {
fn drop(&mut self) {
if let Some(state) = self.state.upgrade() {
state.lock().tasks.remove(&self.id);
}
}
}
impl NetworkTasks {
pub fn current() -> Option<Self> {
CURRENT.try_with(Clone::clone).ok()
}
pub async fn scope<F: Future>(&self, future: F) -> F::Output {
if self.untracked {
future.await
} else {
CURRENT.scope(self.clone(), future).await
}
}
pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
let registration = self.register();
let id = registration.id;
let context = self.clone();
let cancelled = self.cancelled.clone();
let task = self.tracker.spawn(CURRENT.scope(context, async move {
let _registration = registration;
if cancelled.is_cancelled() {
std::future::pending::<()>().await;
}
future.await
}));
self.attach(id, task.abort_handle());
task
}
fn register(&self) -> Registration {
let mut state = self.state.lock();
let id = state.next_id;
state.next_id += 1;
state.tasks.insert(id, None);
Registration {
state: Arc::downgrade(&self.state),
id,
}
}
fn attach(&self, id: u64, task: AbortHandle) {
let mut state = self.state.lock();
if state.cancelled {
task.abort();
}
// Spawning into a stopped runtime can synchronously destroy the registration.
if let Some(slot) = state.tasks.get_mut(&id) {
*slot = Some(task);
}
}
/// Tracks drivers that must process endpoint closure rather than being aborted.
pub fn spawn_completion<F>(&self, future: F)
where
F: Future<Output = ()> + Send + 'static,
{
self.tracker.spawn(CURRENT.scope(self.clone(), future));
}
pub fn cancel(&self) {
self.cancelled.cancel();
let mut state = self.state.lock();
state.cancelled = true;
for task in state.tasks.values().flatten() {
task.abort();
}
self.tracker.close();
}
pub async fn wait(&self) {
self.tracker.wait().await;
}
}
pub(crate) fn spawn<F>(future: F) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
match NetworkTasks::current() {
Some(context) => context.spawn(future),
None => tokio::spawn(future),
}
}
pub(crate) fn spawn_in<F>(tasks: &mut JoinSet<F::Output>, future: F) -> AbortHandle
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
let Some(context) = NetworkTasks::current() else {
return tasks.spawn(future);
};
let registration = context.register();
let id = registration.id;
let cancelled = context.cancelled.clone();
let task = tasks.spawn(context.tracker.track_future(CURRENT.scope(
context.clone(),
async move {
let _registration = registration;
if cancelled.is_cancelled() {
std::future::pending::<()>().await;
}
future.await
},
)));
context.attach(id, task.clone());
task
}
pub(crate) fn inherit<F: Future>(future: F) -> impl Future<Output = F::Output> {
let context = NetworkTasks::current();
async move {
match context {
Some(context) => context.scope(future).await,
None => future.await,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn unmanaged_work_does_not_register_a_network_lifetime() {
let network = NetworkLifetime::untracked();
let mut task = None;
network
.scope(async {
assert!(NetworkTasks::current().is_none());
task = Some(spawn(async { NetworkTasks::current().is_none() }));
})
.await;
network.cancel();
assert!(task.unwrap().await.unwrap());
assert!(network.state.lock().tasks.is_empty());
}
#[tokio::test]
async fn reset_cancels_descendants_and_rejects_late_spawns() {
let context = NetworkTasks::default();
let (ready, started) = tokio::sync::oneshot::channel();
let owned = Arc::new(());
let retained = owned.clone();
let parent = context.spawn(async move {
spawn(async move {
let _retained = retained;
ready.send(()).unwrap();
std::future::pending::<()>().await;
});
});
parent.await.unwrap();
started.await.unwrap();
context.cancel();
context.wait().await;
assert_eq!(Arc::strong_count(&owned), 1);
assert!(context.spawn(async { 1 }).await.unwrap_err().is_cancelled());
context.wait().await;
assert!(context.state.lock().tasks.is_empty());
assert_eq!(spawn(async { 2 }).await.unwrap(), 2);
}
#[tokio::test]
async fn join_set_cancellation_releases_registrations_and_blocks_late_work() {
let network = NetworkLifetime::default();
let mut tasks = JoinSet::new();
network
.scope(async {
spawn_in(&mut tasks, std::future::pending::<()>());
})
.await;
network.cancel();
network.wait().await;
assert!(tasks.join_next().await.unwrap().unwrap_err().is_cancelled());
network
.scope(async {
spawn_in(&mut tasks, async { panic!("retired work ran") });
})
.await;
network.wait().await;
assert!(tasks.join_next().await.unwrap().unwrap_err().is_cancelled());
assert!(network.state.lock().tasks.is_empty());
}
#[test]
fn stopped_runtime_destroys_registrations_without_locking_recursively() {
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
let handle = runtime.handle().clone();
drop(runtime);
let _entered = handle.enter();
let network = NetworkTasks::default();
drop(network.spawn(async {}));
assert!(network.state.lock().tasks.is_empty());
}
}
+249 -21
View File
@@ -153,7 +153,7 @@ async fn prepare(paths: &[String], previous: Option<&DnsRegistry>) -> io::Result
None,
&mut |_, _, _| {},
&mut None,
&crate::desktop::NetworkSession::default(),
&crate::desktop::NetworkSession::default().topology(),
)
.await
}
@@ -165,7 +165,7 @@ async fn prepare_tracking(
previous_servers: Option<&PreparedServers>,
discovered: &mut DiscoverDependencies<'_>,
control: &mut Option<crate::control::runtime::ControlService>,
network: &crate::desktop::NetworkSession,
network: &crate::desktop::NetworkTopology,
) -> io::Result<PreparedServers> {
let mut cache_lease = previous_servers.and_then(|previous| previous.cache_lease.clone());
let (mut validated, snapshot, memory) = load_candidate(
@@ -264,7 +264,7 @@ async fn prepare_tracking(
dns::check_interception(&validated.configs, &validated.dns_groups).await?;
let protect_tun_endpoints = config::protects_tun_endpoints(&validated.configs);
let mut dns = dns::build_dns_registry_for_runtime(
std::mem::take(&mut validated.dns_groups),
validated.dns_groups.clone(),
validated.global_limits,
previous,
&synthetic_endpoints,
@@ -366,6 +366,75 @@ struct RunningServers {
}
impl RunningServers {
async fn pause_network(
&mut self,
prepared: &mut PreparedServers,
control: Option<&crate::control::runtime::ControlService>,
lifetime: &crate::network_tasks::NetworkTasks,
) -> io::Result<()> {
if let (Some(control), Some(generation)) = (control, &prepared.generation) {
control.runtime.begin_activation(generation).await?;
}
#[cfg(unix)]
self.wireguard.cancel();
lifetime.cancel();
self.stop().await;
lifetime.wait().await;
prepared.tcp.defer_admission();
prepared.dns = prepared.dns.retained_state();
self.quic_retirements.wait().await
}
async fn reconnect(
&mut self,
prepared: &mut PreparedServers,
network: &mut crate::desktop::NetworkSession,
) -> io::Result<()> {
self.rebuild_network(prepared, network).await?;
self.start_retry(prepared).await?;
prepared.certificates.start().await;
Ok(())
}
async fn rebuild_network(
&mut self,
prepared: &mut PreparedServers,
network: &mut crate::desktop::NetworkSession,
) -> io::Result<()> {
network.refresh_bootstrap()?;
let endpoints: Vec<_> =
crate::config::tun_endpoint_ips(&prepared.validated.configs).collect();
prepared.dns = dns::build_dns_registry_for_runtime(
prepared.validated.dns_groups.clone(),
prepared.validated.global_limits,
Some(&prepared.dns),
&endpoints,
config::protects_tun_endpoints(&prepared.validated.configs),
)
.await?;
#[cfg(unix)]
match tokio::time::timeout(Duration::from_secs(3), prepared.wireguard.refresh_peers()).await
{
Ok(Ok(())) => {}
Ok(Err(error)) => {
log::warn!(
"WireGuard peer refresh incomplete; retaining unresolved peer endpoints: {error}"
)
}
Err(_) => {
log::warn!("WireGuard peer refresh timed out; retaining unresolved peer endpoints")
}
}
if let Some(generation) = &prepared.generation {
generation.probes.reset_network(
prepared.dns.get_or_create_default(),
prepared.dns.forbidden_ranges().to_vec(),
);
generation.catalog.selection.automatic.choices_changed();
}
Ok(())
}
async fn shutdown_network(
&mut self,
registries: &[&PreparedServers],
@@ -829,14 +898,21 @@ async fn run_inner(
_ = shutdown.recv() => return Ok(()),
result = config::load_config_files(&paths) => result?,
};
*network = crate::desktop::NetworkSession::prepare(&configs).await?;
let mut lifetime = crate::network_tasks::NetworkLifetime::default();
*network = lifetime
.scope(crate::desktop::NetworkSession::prepare(&configs))
.await?;
if !network.monitors_underlay() {
lifetime = crate::network_tasks::NetworkLifetime::untracked();
}
let mut discovered = |paths: &[PathBuf], recovery: &[PathBuf], root: Option<&Path>| {
watches.discovered(paths, recovery, root)
};
let topology = network.topology();
let mut prepared = tokio::select! {
biased;
_ = shutdown.recv() => return Ok(()),
result = prepare_tracking(&paths, configs, None, None, &mut discovered, control, network) => result?,
result = lifetime.scope(prepare_tracking(&paths, configs, None, None, &mut discovered, control, &topology)) => result?,
};
network.prepare_network().await?;
#[cfg(target_os = "linux")]
@@ -857,7 +933,7 @@ async fn run_inner(
tokio::select! {
biased;
_ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await,
result = running.start(&mut prepared) => if let Err(error) = result {
result = lifetime.scope(running.start(&mut prepared)) => if let Err(error) = result {
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
},
@@ -879,11 +955,17 @@ async fn run_inner(
)));
}
println!("Servers ready");
let mut reconnecting = false;
let mut pending_reload = false;
let mut force_network_reset = false;
let mut retry_at = tokio::time::Instant::now();
loop {
let can_retry = reconnecting && network.available();
enum Change {
Shutdown,
TunFailure(io::Error),
Reload { debounce: bool },
Reconnect,
}
let change = wait_with_reconciliation(
async {
@@ -891,21 +973,112 @@ async fn run_inner(
biased;
_ = shutdown.recv() => Change::Shutdown,
error = running.tun_failure() => Change::TunFailure(error),
_ = std::future::ready(()), if force_network_reset => Change::Reconnect,
_ = reload.recv() => Change::Reload { debounce: false },
_ = remote_rule_change(&mut prepared) => Change::Reload { debounce: false },
_ = lifetime.scope(remote_rule_change(&mut prepared)), if !reconnecting => Change::Reload { debounce: false },
_ = watches.changed(&mut changes) => Change::Reload { debounce: true },
_ = tokio::time::sleep_until(retry_at), if can_retry => Change::Reconnect,
_ = std::future::ready(()), if pending_reload && !reconnecting => Change::Reload { debounce: false },
}
},
network,
)
.await;
let debounce = match change {
Ok(Change::Shutdown) => return running.shutdown_network(&[&prepared], network).await,
Ok(Change::TunFailure(error)) => {
Ok(Some(Change::Shutdown)) => {
return running.shutdown_network(&[&prepared], network).await;
}
Ok(Some(Change::TunFailure(error))) => {
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
}
Ok(Change::Reload { debounce }) => debounce,
Ok(Some(Change::Reload { debounce })) => {
if reconnecting {
pending_reload = true;
force_network_reset = network.available();
continue;
}
pending_reload = false;
debounce
}
Ok(None) | Ok(Some(Change::Reconnect)) => {
force_network_reset = false;
if !reconnecting {
log::info!("Underlay changed; retiring network workers");
}
// A rejected recovery reload may have rebuilt transports without starting workers.
if let Err(error) = running
.pause_network(&mut prepared, control.as_ref(), &lifetime)
.await
{
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
}
if !reconnecting && !network.available() {
println!("Network unavailable; waiting for underlay");
}
reconnecting = true;
if !network.available() {
continue;
}
lifetime = crate::network_tasks::NetworkLifetime::default();
let mut result = tokio::select! {
biased;
_ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await,
result = tokio::time::timeout(Duration::from_secs(10), lifetime.scope(async {
if pending_reload {
// Restore rollback state without requiring the old listeners to bind.
running.rebuild_network(&mut prepared, network).await
} else {
running.reconnect(&mut prepared, network).await
}
})) => result.unwrap_or_else(|_| Err(io::Error::new(io::ErrorKind::TimedOut, "network recovery timed out"))),
};
// Ownership failures remain terminal, unlike transient worker startup errors.
match network.reconcile().await {
Err(error) => {
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
}
Ok(changed) if changed || !network.available() => {
result = Err(io::Error::new(
io::ErrorKind::Interrupted,
"underlay changed during recovery",
));
}
Ok(_) => {}
}
match result {
Ok(()) if pending_reload => {
pending_reload = false;
false
}
Ok(()) => {
if let Some(control) = control.as_ref()
&& let Err(error) = control.runtime.resume().await
{
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
}
prepared.tcp.activate();
reconnecting = false;
println!("Network reconnected");
continue;
}
Err(error) => {
log::warn!("Network recovery deferred: {error}");
if let Err(error) = running
.pause_network(&mut prepared, control.as_ref(), &lifetime)
.await
{
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
}
retry_at = tokio::time::Instant::now() + Duration::from_secs(2);
continue;
}
}
}
Err(error) => {
let cleanup = running.shutdown_network(&[&prepared], network).await;
return Err(io::Error::other(format!(
@@ -914,15 +1087,31 @@ async fn run_inner(
}
};
if debounce {
tokio::select! {
let settled = tokio::select! {
biased;
_ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await,
error = running.tun_failure() => {
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
},
_ = reload.recv() => {}
_ = watches.debounce(&mut changes) => {}
result = wait_with_reconciliation(async {
tokio::select! {
_ = reload.recv() => {},
_ = watches.debounce(&mut changes) => {},
}
}, network) => result,
};
match settled {
Ok(Some(())) => {}
Ok(None) => {
pending_reload = true;
force_network_reset = true;
continue;
}
Err(error) => {
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
}
}
}
while changes.try_recv().is_ok() {}
@@ -934,6 +1123,7 @@ async fn run_inner(
let mut discovered = |paths: &[PathBuf], recovery: &[PathBuf], root: Option<&Path>| {
watches.discovered(paths, recovery, root)
};
let topology = network.topology();
let candidate = tokio::select! {
biased;
_ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await,
@@ -941,10 +1131,22 @@ async fn run_inner(
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
},
result = async {
result = wait_with_reconciliation(async {
let configs = config::load_config_files(&paths).await?;
prepare_tracking(&paths, configs, Some(&prepared.dns), Some(&prepared), &mut discovered, control, network).await
} => result,
lifetime.scope(prepare_tracking(&paths, configs, Some(&prepared.dns), Some(&prepared), &mut discovered, control, &topology)).await
}, network) => result,
};
let candidate = match candidate {
Ok(Some(candidate)) => candidate,
Ok(None) => {
pending_reload = true;
force_network_reset = true;
continue;
}
Err(error) => {
let _ = running.shutdown_network(&[&prepared], network).await;
return Err(error);
}
};
let mut candidate = match candidate {
Ok(candidate) => candidate,
@@ -993,15 +1195,31 @@ async fn run_inner(
let result = tokio::select! {
biased;
_ = shutdown.recv() => return running.shutdown_network(&[&prepared, &candidate], network).await,
result = running.activate(&mut candidate, control.as_ref()) => result,
result = lifetime.scope(running.activate(&mut candidate, control.as_ref())) => result,
};
if let Err(error) = result {
let rollback = tokio::select! {
biased;
_ = shutdown.recv() => return running.shutdown_network(&[&prepared, &candidate], network).await,
result = running.rollback(&mut prepared, &mut candidate, control.as_ref()) => result,
result = lifetime.scope(running.rollback(&mut prepared, &mut candidate, control.as_ref())) => result,
};
if let Err(rollback) = rollback {
if reconnecting {
eprintln!(
"Reload rejected; previous servers unavailable, recovery continues: {error}; rollback: {rollback}"
);
if let Err(error) = running
.pause_network(&mut prepared, control.as_ref(), &lifetime)
.await
{
let _ = running
.shutdown_network(&[&prepared, &candidate], network)
.await;
return Err(error);
}
retry_at = tokio::time::Instant::now() + Duration::from_secs(2);
continue;
}
let cleanup = running
.shutdown_network(&[&prepared, &candidate], network)
.await;
@@ -1010,11 +1228,19 @@ async fn run_inner(
)));
}
eprintln!("Reload rejected; previous servers restored: {error}");
if reconnecting {
reconnecting = false;
println!("Network reconnected");
}
continue;
}
watches.activated(&candidate.validated.reload_files);
let mut retired = std::mem::replace(&mut prepared, candidate);
retired.certificates.shutdown().await;
if reconnecting {
reconnecting = false;
println!("Network reconnected");
}
println!("Servers ready");
}
}
@@ -1022,15 +1248,17 @@ async fn run_inner(
async fn wait_with_reconciliation<T>(
change: impl std::future::Future<Output = T>,
network: &mut crate::desktop::NetworkSession,
) -> io::Result<T> {
) -> io::Result<Option<T>> {
tokio::pin!(change);
loop {
tokio::select! {
biased;
result = &mut change => return Ok(result),
result = &mut change => return Ok(Some(result)),
_ = network.changed() => {
// Retain in-flight refreshes and finish network mutations before handling reload/shutdown.
network.reconcile().await?;
if network.reconcile().await? {
return Ok(None);
}
}
}
}
+353 -3
View File
@@ -37,7 +37,7 @@ async fn ntp_initial_sync_precedes_deferred_certificate_binds() {
Some(&previous),
&mut |_, _, _| {},
&mut None,
&crate::desktop::NetworkSession::default(),
&crate::desktop::NetworkSession::default().topology(),
)
.await
.unwrap();
@@ -98,7 +98,7 @@ async fn network_reconciliation_does_not_cancel_inflight_refresh() {
request.push(stream.read_u8().await.unwrap());
}
for _ in 0..3 {
events.send(()).await.unwrap();
events.send(None).await.unwrap();
reconciled.recv().await.unwrap();
}
stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes()).await.unwrap();
@@ -106,13 +106,363 @@ async fn network_reconciliation_does_not_cancel_inflight_refresh() {
std::future::pending::<()>().await;
};
tokio::select! {
result = wait_with_reconciliation(remote_rule_change(&mut prepared), &mut network) => result.unwrap(),
result = wait_with_reconciliation(remote_rule_change(&mut prepared), &mut network) => result.unwrap().unwrap(),
_ = fixture => unreachable!(),
}
assert_eq!(std::fs::read(path).unwrap(), body);
}).await.unwrap();
}
#[tokio::test]
async fn underlay_change_interrupts_the_wait_without_consuming_a_reload() {
let (events, receiver) = mpsc::channel(1);
let (ack, _reconciled) = mpsc::channel(1);
let mut network = crate::desktop::NetworkSession::with_test_events(receiver, ack);
events.send(Some(true)).await.unwrap();
assert!(
wait_with_reconciliation(std::future::pending::<()>(), &mut network)
.await
.unwrap()
.is_none()
);
}
#[cfg(unix)]
#[tokio::test]
async fn reconnect_loop_defers_offline_reload_and_accepts_repairs_after_failed_start() {
const NAME: &str = "process::tests::reconnect_loop_defers_offline_reload_and_accepts_repairs_after_failed_start";
const CHILD: &str = "SHOES_RECONNECT_LOOP_TEST";
if std::env::var_os(CHILD).is_none() {
let output = std::process::Command::new(std::env::current_exe().unwrap())
.args(["--exact", NAME, "--nocapture"])
.env(CHILD, "1")
.output()
.unwrap();
shoes_test_support::isolated::assert_one_test_passed(&output);
assert!(
String::from_utf8_lossy(&output.stderr)
.contains("Reload rejected; previous servers unavailable, recovery continues:")
);
return;
}
use tokio::io::{AsyncReadExt, AsyncWriteExt};
async fn greet(address: std::net::SocketAddr) -> tokio::net::TcpStream {
loop {
if let Ok(mut stream) = tokio::net::TcpStream::connect(address).await {
stream.write_all(&[5, 1, 0]).await.unwrap();
let mut reply = [0; 2];
stream.read_exact(&mut reply).await.unwrap();
assert_eq!(reply, [5, 0]);
return stream;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
}
tokio::time::timeout(Duration::from_secs(20), async {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("config.json");
let unix = directory.path().join("listener.sock");
let first = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let second = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = first.local_addr().unwrap();
let added = second.local_addr().unwrap();
let mut config = vec![
serde_json::json!({"address":address.to_string(), "protocol":{"type":"socks"}}),
serde_json::json!({"path":unix, "protocol":{"type":"socks"}}),
];
std::fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap();
drop((first, second));
let (events, receiver) = mpsc::channel(1);
let (ack, mut reconciled) = mpsc::channel(1);
let network = crate::desktop::NetworkSession::with_test_events(receiver, ack);
let process = crate::desktop::TEST_NETWORK.scope(
std::cell::RefCell::new(Some(network)),
Box::pin(run(vec![path.to_string_lossy().into_owned()], false, true)),
);
tokio::pin!(process);
let fixture = async {
let mut live = greet(address).await;
events.send(Some(false)).await.unwrap();
reconciled.recv().await.unwrap();
assert_eq!(live.read(&mut [0]).await.unwrap(), 0);
config.push(
serde_json::json!({"address":added.to_string(), "protocol":{"type":"socks"}}),
);
std::fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap();
assert_eq!(
unsafe { libc::kill(std::process::id() as i32, libc::SIGHUP) },
0
);
tokio::time::sleep(Duration::from_millis(100)).await;
assert!(tokio::net::TcpStream::connect(added).await.is_err());
events.send(Some(true)).await.unwrap();
let mut live = greet(added).await;
events.send(Some(false)).await.unwrap();
assert_eq!(live.read(&mut [0]).await.unwrap(), 0);
std::fs::remove_file(&unix).unwrap();
std::fs::create_dir(&unix).unwrap();
events.send(Some(true)).await.unwrap();
// A non-AddrInUse startup error must keep retrying while reload remains usable.
tokio::time::sleep(Duration::from_millis(2300)).await;
let mut queued = tokio::net::TcpStream::connect(address).await.unwrap();
queued.write_all(&[5, 1, 0]).await.unwrap();
assert!(
tokio::time::timeout(Duration::from_millis(100), queued.read_u8())
.await
.is_err()
);
assert_eq!(
unsafe { libc::kill(std::process::id() as i32, libc::SIGHUP) },
0
);
// An unchanged reload fails activation and rollback, but must not end recovery.
assert!(
tokio::time::timeout(Duration::from_secs(1), queued.read_u8())
.await
.is_err()
);
config.remove(1);
std::fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap();
assert_eq!(
unsafe { libc::kill(std::process::id() as i32, libc::SIGHUP) },
0
);
assert_eq!(queued.read_u8().await.unwrap(), 5);
assert_eq!(queued.read_u8().await.unwrap(), 0);
assert!(unix.is_dir());
let mut live = greet(added).await;
events.send(Some(false)).await.unwrap();
assert_eq!(live.read(&mut [0]).await.unwrap(), 0);
assert_eq!(
unsafe { libc::kill(std::process::id() as i32, libc::SIGTERM) },
0
);
};
tokio::select! {
result = &mut process => panic!("process exited before recovery: {result:?}"),
_ = fixture => {},
}
process.await.unwrap();
})
.await
.unwrap();
}
#[tokio::test]
async fn reconnect_restarts_certificate_workers_retired_by_a_rejected_candidate() {
const NAME: &str =
"process::tests::reconnect_restarts_certificate_workers_retired_by_a_rejected_candidate";
const CHILD: &str = "SHOES_RECONNECT_CERTIFICATE_TEST";
if std::env::var_os(CHILD).is_none() {
shoes_test_support::isolated::run_isolated(NAME, CHILD, "1");
return;
}
use tokio::io::{AsyncReadExt, AsyncWriteExt};
async fn challenge(address: std::net::SocketAddr) {
tokio::time::timeout(Duration::from_secs(2), async {
let mut stream = tokio::net::TcpStream::connect(address).await.unwrap();
stream
.write_all(b"GET / HTTP/1.1\r\nHost: proxy.test\r\n\r\n")
.await
.unwrap();
let mut response = String::new();
stream.read_to_string(&mut response).await.unwrap();
assert!(response.starts_with("HTTP/1.1 404 "));
})
.await
.unwrap();
}
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("config.json");
let reserved = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = reserved.local_addr().unwrap();
std::fs::write(&path, serde_json::json!([
{"certificate_source":"managed", "type":"acme", "domains":["proxy.test"],
"directory_url":"http://127.0.0.1:9/directory", "state_dir":directory.path().join("state"),
"challenge":{"type":"http01", "listen":[address.to_string()]}},
{"address":"127.0.0.1:0", "protocol":{"type":"socks"}}
]).to_string()).unwrap();
drop(reserved);
let old = crate::network_tasks::NetworkLifetime::default();
let mut prepared = old
.scope(prepare(&[path.to_string_lossy().into_owned()], None))
.await
.unwrap();
let mut running = RunningServers::default();
old.scope(running.activate(&mut prepared, None))
.await
.unwrap();
challenge(address).await;
// Candidate retirement can stop the accepted certificate workers before rollback fails.
prepared
.certificates
.retire_unused(&Default::default())
.await;
running
.pause_network(&mut prepared, None, &old)
.await
.unwrap();
let fresh = crate::network_tasks::NetworkLifetime::default();
fresh
.scope(running.reconnect(
&mut prepared,
&mut crate::desktop::NetworkSession::default(),
))
.await
.unwrap();
challenge(address).await;
running.shutdown(&[&prepared]).await.unwrap();
}
#[tokio::test]
async fn network_reset_keeps_controller_generation_and_explicit_selection() {
const CHILD: &str = "SHOES_RECONNECT_CONTROL_TEST";
if std::env::var_os(CHILD).is_none() {
shoes_test_support::isolated::run_isolated(
"process::tests::network_reset_keeps_controller_generation_and_explicit_selection",
CHILD,
"1",
);
return;
}
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("config.json");
std::fs::write(
&path,
serde_json::json!([
{"client_group":"chosen", "type":"selector", "client_chains":[
{"chain":[{"name":"first", "protocol":{"type":"direct"}}]},
{"chain":[{"name":"second", "protocol":{"type":"direct"}}]}
]},
{"address":"127.0.0.1:0", "protocol":{"type":"socks"}}
])
.to_string(),
)
.unwrap();
let paths = [path.to_string_lossy().into_owned()];
let configs = config::load_config_files(&paths).await.unwrap();
let mut control = None;
let mut network = crate::desktop::NetworkSession::default();
let old = crate::network_tasks::NetworkLifetime::default();
let mut prepared = old
.scope(prepare_tracking(
&paths,
configs,
None,
None,
&mut |_, _, _| {},
&mut control,
&network.topology(),
))
.await
.unwrap();
let mut control = control.unwrap();
let generation = prepared.generation.clone().unwrap();
control.runtime.publish(generation.clone()).await.unwrap();
control.runtime.resume().await.unwrap();
control
.runtime
.select(generation.id, "chosen".into(), "second".into())
.await
.unwrap();
let selection = generation.catalog.selection.snapshot();
let mut running = RunningServers::default();
old.scope(running.start(&mut prepared)).await.unwrap();
running
.pause_network(&mut prepared, Some(&control), &old)
.await
.unwrap();
let fresh = crate::network_tasks::NetworkLifetime::default();
fresh
.scope(running.reconnect(&mut prepared, &mut network))
.await
.unwrap();
control.runtime.resume().await.unwrap();
prepared.tcp.activate();
assert!(std::sync::Arc::ptr_eq(
&generation,
&control.runtime.generation().unwrap()
));
assert!(std::sync::Arc::ptr_eq(
&selection,
&generation.catalog.selection.snapshot()
));
assert_eq!(selection["chosen"], "second");
running
.pause_network(&mut prepared, Some(&control), &fresh)
.await
.unwrap();
running.shutdown(&[&prepared]).await.unwrap();
control.shutdown().await;
}
#[tokio::test]
async fn reconnect_uses_memory_and_cancels_current_and_draining_connections() {
const NAME: &str =
"process::tests::reconnect_uses_memory_and_cancels_current_and_draining_connections";
const CHILD: &str = "SHOES_RECONNECT_PROCESS_TEST";
if std::env::var_os(CHILD).is_none() {
shoes_test_support::isolated::run_isolated(NAME, CHILD, "1");
return;
}
use tokio::io::{AsyncReadExt, AsyncWriteExt};
tokio::time::timeout(Duration::from_secs(15), async {
let echo = shoes_test_support::test_servers::start_tcp_stream_echo_server("127.0.0.1", 0).await.unwrap();
let reserved = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let listen = reserved.local_addr().unwrap();
let reserved_socks = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let socks = reserved_socks.local_addr().unwrap();
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("config.json");
std::fs::write(&path, serde_json::json!([
{"global_settings": {"reload_grace_secs":300}},
{"address":listen.to_string(), "protocol":{"type":"forward", "target":echo.local_addr().to_string()}},
{"address":socks.to_string(), "protocol":{"type":"socks"}}
]).to_string()).unwrap();
drop((reserved, reserved_socks));
let mut lifetime = crate::network_tasks::NetworkLifetime::default();
let mut prepared = lifetime.scope(prepare(&[path.to_string_lossy().into_owned()], None)).await.unwrap();
let mut running = RunningServers::default();
lifetime.scope(running.start(&mut prepared)).await.unwrap();
let mut draining = tokio::net::TcpStream::connect(listen).await.unwrap();
draining.write_all(b"before").await.unwrap();
let mut bytes = [0; 6];
draining.read_exact(&mut bytes).await.unwrap();
running.stop().await;
lifetime.scope(running.start(&mut prepared)).await.unwrap();
draining.write_all(b"reload").await.unwrap();
draining.read_exact(&mut bytes).await.unwrap();
assert_eq!(&bytes, b"reload");
let mut pending = tokio::net::TcpStream::connect(socks).await.unwrap();
pending.write_all(&[5]).await.unwrap();
let mut current = tokio::net::TcpStream::connect(listen).await.unwrap();
current.write_all(b"active").await.unwrap();
current.read_exact(&mut bytes).await.unwrap();
std::fs::write(&path, "invalid: [").unwrap();
running.pause_network(&mut prepared, None, &lifetime).await.unwrap();
for stream in [&mut draining, &mut pending, &mut current] {
let result = tokio::time::timeout(Duration::from_secs(1), stream.read(&mut [0])).await
.expect("network reset waited for ordinary reload grace");
assert!(matches!(result, Ok(0)) || result.is_err(), "{result:?}");
}
lifetime = crate::network_tasks::NetworkLifetime::default();
lifetime.scope(running.reconnect(&mut prepared, &mut crate::desktop::NetworkSession::default()))
.await.unwrap();
prepared.tcp.activate();
let mut fresh = tokio::net::TcpStream::connect(listen).await.unwrap();
fresh.write_all(b"fresh!").await.unwrap();
fresh.read_exact(&mut bytes).await.unwrap();
assert_eq!(&bytes, b"fresh!");
running.pause_network(&mut prepared, None, &lifetime).await.unwrap();
}).await.unwrap();
}
async fn load_validated(
paths: &[String],
discovered: &mut DiscoverDependencies<'_>,
+11 -9
View File
@@ -74,16 +74,18 @@ impl HoppingSocket {
task: parking_lot::Mutex::new(None),
});
let weak = Arc::downgrade(&socket);
*socket.task.lock() = Some(AbortOnDropHandle::new(tokio::spawn(async move {
loop {
let seconds = rand::rng().random_range(min..=max);
tokio::time::sleep(Duration::from_secs(seconds)).await;
let Some(socket) = weak.upgrade() else { return };
if let Err(error) = socket.hop() {
log::warn!("HY2 port hop failed; keeping current socket: {error}");
*socket.task.lock() = Some(AbortOnDropHandle::new(crate::network_tasks::spawn(
async move {
loop {
let seconds = rand::rng().random_range(min..=max);
tokio::time::sleep(Duration::from_secs(seconds)).await;
let Some(socket) = weak.upgrade() else { return };
if let Err(error) = socket.hop() {
log::warn!("HY2 port hop failed; keeping current socket: {error}");
}
}
}
})));
},
)));
Ok(socket)
}
+1 -1
View File
@@ -79,7 +79,7 @@ pub(super) async fn authenticate(
.await
.map_err(io::Error::other)?;
let hold = sender.clone();
let task = AbortOnDropHandle::new(tokio::spawn(async move {
let task = AbortOnDropHandle::new(crate::network_tasks::spawn(async move {
// Both the driver and at least one SendRequest must survive authentication.
let _hold = hold;
let _ = driver.wait_idle().await;
+23
View File
@@ -84,6 +84,29 @@ fn protocol() -> Protocol {
})
}
#[tokio::test]
async fn network_reset_cancels_warmed_pool_and_retained_connection_handles() {
tokio::time::timeout(Duration::from_secs(5), async {
let network = crate::network_tasks::NetworkLifetime::default();
let (server, dialer) = network.scope(async { endpoints() }).await;
let pool = Pool::default();
let protocol = protocol();
let (session, peer) =
tokio::join!(network.scope(pool.get(dialer, &protocol, true)), async {
server.accept().await.unwrap().await.unwrap()
},);
let session = session.unwrap();
let mut auth = peer.accept_uni().await.unwrap();
assert_eq!(auth.read_to_end(50).await.unwrap().len(), 50);
network.cancel();
network.wait().await;
session.connection().closed().await;
assert!(session.connection().open_bi().await.is_err());
})
.await
.unwrap();
}
#[tokio::test]
async fn exhausted_udp_ids_replace_the_connection_without_migrating_live_associations() {
use crate::tcp::proxy_connector::ProxyConnector;
+19 -16
View File
@@ -71,7 +71,7 @@ impl Pool {
let (sender, result) = watch::channel(None);
let slots = Arc::downgrade(&self.slots);
let key = dialer.route_key().clone();
let task = tokio::spawn(async move {
let task = crate::network_tasks::spawn(async move {
let result = tokio::time::timeout(SETUP_TIMEOUT, Generation::open(dialer, protocol))
.await
.unwrap_or_else(|_| {
@@ -162,28 +162,31 @@ impl Generation {
generation.udp_enabled = config.udp_enabled;
generation
.tasks
.push(AbortOnDropHandle::new(tokio::spawn(tuic::heartbeat(
generation.connection().clone(),
config.heartbeat_interval_secs,
))));
.push(AbortOnDropHandle::new(crate::network_tasks::spawn(
tuic::heartbeat(
generation.connection().clone(),
config.heartbeat_interval_secs,
),
)));
let (sender, receiver) = tokio::sync::mpsc::channel(64);
generation.dissociate = Some(sender);
generation
.tasks
.push(AbortOnDropHandle::new(tokio::spawn(tuic::dissociate_loop(
commands.clone(),
receiver,
))));
.push(AbortOnDropHandle::new(crate::network_tasks::spawn(
tuic::dissociate_loop(commands.clone(), receiver),
)));
generation.commands = Some(commands);
}
}
generation
.tasks
.push(AbortOnDropHandle::new(tokio::spawn(udp::receive(
generation.connection().clone(),
generation.registry.clone(),
matches!(generation.protocol, Protocol::Tuic(_)),
))));
.push(AbortOnDropHandle::new(crate::network_tasks::spawn(
udp::receive(
generation.connection().clone(),
generation.registry.clone(),
matches!(generation.protocol, Protocol::Tuic(_)),
),
)));
generation.initialized = true;
Ok(generation)
}
@@ -198,7 +201,7 @@ impl Drop for Generation {
self.registry.lock().close();
if self.initialized
&& self.connection().close_reason().is_none()
&& let Ok(runtime) = tokio::runtime::Handle::try_current()
&& tokio::runtime::Handle::try_current().is_ok()
{
// Retiring a route must not discard FIN/data already accepted by
// Quinn. One bounded drain task per generation retains its endpoint
@@ -206,7 +209,7 @@ impl Drop for Generation {
let connection = self.connection().clone();
let endpoint = self.carrier._endpoint.clone();
let tasks = std::mem::take(&mut self.tasks);
runtime.spawn(async move {
crate::network_tasks::spawn(async move {
let _ = tokio::time::timeout(Duration::from_secs(3), connection.closed()).await;
connection.close(0u32.into(), b"client retired");
drop((tasks, endpoint));
+1 -1
View File
@@ -46,7 +46,7 @@ impl CommandWriter {
// Dropping a partial command sends FIN, which peers treat as a fatal
// protocol error. Once admitted, finish independently of caller cancellation.
// The permit bounds detached writers; connection closure releases them.
tokio::spawn(async move {
crate::network_tasks::spawn(async move {
let _permit = permit;
stream.write_all(&bytes).await?;
stream.finish()?;
+1 -1
View File
@@ -126,7 +126,7 @@ pub(super) async fn receive(
}
stream = connection.accept_uni(), if is_tuic && streams.len() < QUEUE_SIZE => {
let Ok(mut stream) = stream else { break };
streams.spawn(async move {
crate::network_tasks::spawn_in(&mut streams, async move {
tokio::time::timeout(
Duration::from_secs(5),
stream.read_to_end(tuic::MAX_PACKET),
+56 -8
View File
@@ -19,7 +19,8 @@ const LISTENER_CLOSE_GRACE: Duration = Duration::from_secs(1);
#[derive(Debug)]
struct ListenerRuntime {
endpoint_started: AtomicBool,
retired: CancellationToken,
retired: Option<CancellationToken>,
network: Option<crate::network_tasks::NetworkTasks>,
}
impl Runtime for ListenerRuntime {
@@ -31,12 +32,29 @@ impl Runtime for ListenerRuntime {
// Quinn 0.11 spawns the endpoint driver synchronously before exposing the
// endpoint. Later spawns are connection drivers; recheck this on upgrades.
if self.endpoint_started.swap(true, Ordering::Relaxed) {
quinn::TokioRuntime.spawn(future);
match &self.network {
Some(network) => {
network.spawn_completion(future);
}
None => quinn::TokioRuntime.spawn(future),
}
return;
}
let mut driver = AbortOnDropHandle::new(tokio::spawn(future));
let retired = self.retired.clone();
tokio::spawn(async move {
if self.retired.is_none() {
match &self.network {
Some(network) => {
network.spawn(future);
}
None => quinn::TokioRuntime.spawn(future),
}
return;
}
let mut driver = AbortOnDropHandle::new(match &self.network {
Some(network) => network.spawn(future),
None => tokio::spawn(future),
});
let retired = self.retired.clone().unwrap();
let supervisor = async move {
tokio::select! {
_ = &mut driver => {},
_ = async {
@@ -46,7 +64,15 @@ impl Runtime for ListenerRuntime {
}
// Dropping the endpoint driver closes its connection-event channels.
// Connection drivers then terminate and wake application waiters.
});
};
match &self.network {
Some(network) => {
network.spawn(supervisor);
}
None => {
tokio::spawn(supervisor);
}
}
}
fn wrap_udp_socket(&self, socket: std::net::UdpSocket) -> io::Result<Arc<dyn AsyncUdpSocket>> {
@@ -116,15 +142,24 @@ impl QuicEndpoint {
let retired = CancellationToken::new();
let runtime = ListenerRuntime {
endpoint_started: AtomicBool::new(false),
retired: retired.clone(),
retired: Some(retired.clone()),
network: crate::network_tasks::NetworkTasks::current(),
};
(Arc::new(runtime), Some(retired.drop_guard()))
} else {
(Arc::new(quinn::TokioRuntime), None)
(
Arc::new(ListenerRuntime {
endpoint_started: AtomicBool::new(false),
retired: None,
network: crate::network_tasks::NetworkTasks::current(),
}),
None,
)
};
let socket = MemorySocket {
inner: socket,
endpoint_memory: None,
cancelled: None,
_release: release,
};
quinn::Endpoint::new_with_abstract_socket(
@@ -222,6 +257,9 @@ pub(crate) fn socket_with_memory(
Ok(Arc::new(MemorySocket {
inner: quinn::TokioRuntime.wrap_udp_socket(socket)?,
endpoint_memory: Some(memory),
// Hickory owns its Quinn runtime; retire its driver through the socket instead.
cancelled: crate::network_tasks::NetworkTasks::current()
.map(|network| parking_lot::Mutex::new(Box::pin(network.cancelled.cancelled_owned()))),
_release: None,
}))
}
@@ -230,6 +268,8 @@ pub(crate) fn socket_with_memory(
struct MemorySocket {
inner: Arc<dyn AsyncUdpSocket>,
endpoint_memory: Option<BudgetPermit>,
cancelled:
Option<parking_lot::Mutex<Pin<Box<tokio_util::sync::WaitForCancellationFutureOwned>>>>,
// Field order signals retirement only after the native socket has been dropped.
_release: Option<oneshot::Sender<()>>,
}
@@ -265,6 +305,14 @@ impl AsyncUdpSocket for MemorySocket {
bufs: &mut [IoSliceMut<'_>],
meta: &mut [quinn::udp::RecvMeta],
) -> Poll<io::Result<usize>> {
if let Some(cancelled) = &self.cancelled
&& cancelled.lock().as_mut().poll(cx).is_ready()
{
return Poll::Ready(Err(io::Error::new(
io::ErrorKind::Interrupted,
"DNS QUIC network retired",
)));
}
self.inner.poll_recv(cx, bufs, meta)
}
+57 -2
View File
@@ -11,7 +11,8 @@ async fn listener_runtime_cancels_only_the_endpoint_driver_after_grace() {
let retired = CancellationToken::new();
let runtime = ListenerRuntime {
endpoint_started: AtomicBool::new(false),
retired: retired.clone(),
retired: Some(retired.clone()),
network: None,
};
let endpoint = Arc::new(());
let owned_endpoint = endpoint.clone();
@@ -48,7 +49,8 @@ async fn listener_runtime_reaps_a_completed_driver_and_its_watcher() {
let baseline = metrics.num_alive_tasks();
let runtime = ListenerRuntime {
endpoint_started: AtomicBool::new(false),
retired: CancellationToken::new(),
retired: Some(CancellationToken::new()),
network: None,
};
runtime.spawn(Box::pin(async {}));
for _ in 0..4 {
@@ -57,6 +59,58 @@ async fn listener_runtime_reaps_a_completed_driver_and_its_watcher() {
assert_eq!(metrics.num_alive_tasks(), baseline);
}
#[tokio::test]
async fn network_reset_retires_external_quinn_runtime_through_dns_socket() {
timeout(Duration::from_secs(3), async {
let network = crate::network_tasks::NetworkLifetime::default();
let (client_config, server_config) = configs();
let server =
quinn::Endpoint::server(server_config, "127.0.0.1:0".parse().unwrap()).unwrap();
let budget = Arc::new(Budget::new(Some(1)));
let socket = network
.scope(async {
socket_with_memory(
std::net::UdpSocket::bind("127.0.0.1:0").unwrap(),
budget.acquire(1).unwrap(),
)
.unwrap()
})
.await;
let mut client = quinn::Endpoint::new_with_abstract_socket(
quinn::EndpointConfig::default(),
None,
socket,
Arc::new(quinn::TokioRuntime),
)
.unwrap();
client.set_default_client_config(client_config);
let (connection, peer) = tokio::join!(
async {
client
.connect(server.local_addr().unwrap(), "localhost")
.unwrap()
.await
.unwrap()
},
async { server.accept().await.unwrap().await.unwrap() },
);
connection
.send_datagram(Bytes::from_static(b"before"))
.unwrap();
assert_eq!(peer.read_datagram().await.unwrap(), b"before"[..]);
network.cancel();
connection.closed().await;
assert!(
connection
.send_datagram(Bytes::from_static(b"after"))
.is_err()
);
drop((connection, client, peer, server));
})
.await
.unwrap();
}
#[tokio::test]
async fn endpoint_cancellation_wakes_retained_handles_before_releasing_memory() {
let (client_config, server_config) = configs();
@@ -119,6 +173,7 @@ async fn socket_release_waits_for_delegated_poller() {
let socket = Arc::new(MemorySocket {
inner: quinn::TokioRuntime.wrap_udp_socket(socket).unwrap(),
endpoint_memory: None,
cancelled: None,
_release: Some(release),
});
let budget = Arc::new(Budget::new(Some(1)));
+3 -3
View File
@@ -34,7 +34,7 @@ async fn start_quic_server(
)? {
let resolver = resolver.clone();
let server_handler = server_handler.clone();
let join_handle = tokio::spawn(async move {
let join_handle = crate::network_tasks::spawn(async move {
let mut tasks = crate::listener_tasks::ListenerTasks::immediate();
loop {
let conn = tokio::select! {
@@ -104,7 +104,7 @@ async fn process_connection(
let cloned_resolver = resolver.clone();
let cloned_handler = server_handler.clone();
let peer = connection.remote_address();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
if let Err(e) = process_streams(cloned_resolver, cloned_handler, stream, peer).await {
error!("Failed to process streams: {e}");
}
@@ -260,7 +260,7 @@ pub async fn start_quic_servers(
}
for endpoint in endpoints {
let service = service.clone();
handles.push(tokio::spawn(async move {
handles.push(crate::network_tasks::spawn(async move {
let mut tasks = crate::listener_tasks::ListenerTasks::immediate();
loop {
let incoming = tokio::select! {
+28 -1
View File
@@ -44,7 +44,7 @@ impl UdpRelay {
output,
peer,
};
let task = tokio::spawn(async move {
let task = crate::network_tasks::spawn(async move {
let _ = run_udp_routing(
ServerStream::Targeted(Box::new(stream)),
selector,
@@ -200,6 +200,33 @@ mod tests {
}
}
#[tokio::test]
async fn network_reset_retires_relay_even_while_its_handle_survives() {
let network = crate::network_tasks::NetworkLifetime::default();
let resolver: Arc<dyn Resolver> = Arc::new(NoHostnameResolver);
let selector = Arc::new(ClientProxySelector::new(vec![ConnectRule::new(
vec![crate::address::NetLocationMask::from("0.0.0.0/0").unwrap()],
ConnectAction::new_allow(
None,
crate::tcp::chain_builder::build_direct_chain_group(resolver.clone()),
),
)]));
let relay = network
.scope(async { UdpRelay::new(selector, resolver, None).unwrap() })
.await;
network.cancel();
network.wait().await;
assert!(relay.recv().await.is_err());
assert!(
relay
.send_to(
Bytes::new(),
NetLocation::from_str("192.0.2.1:53", None).unwrap()
)
.is_err()
);
}
#[tokio::test]
async fn owned_input_preserves_allocation_and_queue_drop_policy() {
let (tx, mut input) = mpsc::channel(QUEUE_SIZE);
+15
View File
@@ -22,6 +22,21 @@ pub(crate) struct SocketPolicy(());
#[cfg(target_os = "linux")]
impl SocketPolicy {
pub(crate) fn refresh_bootstrap(&self) -> io::Result<()> {
let endpoints = bootstrap_endpoints()
.ok_or_else(|| io::Error::other("managed socket policy disappeared"))?;
let bootstrap = crate::dns::controlled_bootstrap(&endpoints)?;
let old = {
let mut current = POLICY.write();
let policy = current
.as_mut()
.ok_or_else(|| io::Error::other("managed socket policy disappeared"))?;
std::mem::replace(&mut policy.bootstrap, bootstrap)
};
drop(old);
Ok(())
}
pub(crate) fn install(
interface: String,
bypass_mark: u32,
+3 -3
View File
@@ -556,7 +556,7 @@ pub(crate) async fn start_servers_with_listeners(
let task = tokio_util::task::AbortOnDropHandle::new(
start_tun_server(tun_config, resolver).await?,
);
Ok(vec![tokio::spawn(async move {
Ok(vec![crate::network_tasks::spawn(async move {
if let Err(error) = task
.await
.unwrap_or_else(|error| Err(std::io::Error::other(error)))
@@ -671,7 +671,7 @@ async fn start_tcp_servers(
let tcp_config = tcp_config.clone();
let resolver = resolver.clone();
let admission = sockets.admission.clone();
handles.push(tokio::spawn(async move {
handles.push(crate::network_tasks::spawn(async move {
// Uncommitted flows must not outlive rollback through graceful draining.
if let Some(admission) = admission {
admission.cancelled().await;
@@ -699,7 +699,7 @@ async fn start_tcp_servers(
create_tcp_server_handler(protocol, &client_proxy_selector, &resolver, None)?
.into();
let admission = sockets.admission.clone();
let handle = tokio::spawn(async move {
let handle = crate::network_tasks::spawn(async move {
if let Some(admission) = admission {
admission.cancelled().await;
}
+5 -5
View File
@@ -262,7 +262,7 @@ async fn run_bidirectional_loop(
};
let client_proxy_selector = client_proxy_selector.clone();
let resolver = resolver.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
let _permit = permit;
let result = async {
let Some(stream) = reader.await? else {
@@ -528,7 +528,7 @@ impl UdpSession {
};
session.task = Some(
tokio::spawn(async move {
crate::network_tasks::spawn(async move {
let result = if is_uni_stream {
run_udp_remote_to_local_stream_loop(
assoc_id,
@@ -717,7 +717,7 @@ async fn run_unidirectional_loop(
// Spawn a cleanup task for UDP sessions that terminates when connection closes
let cleanup_session_map = udp_session_map.clone();
let cleanup_cancel_token = cancel_token.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
let mut interval = tokio::time::interval(Duration::from_secs(1));
loop {
tokio::select! {
@@ -781,7 +781,7 @@ async fn run_unidirectional_loop(
let resolver = resolver.clone();
let udp_session_map = udp_session_map.clone();
let cancel_token = cancel_token.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
let _permit = permit;
// Per TUIC protocol, each uni stream carries exactly ONE command.
// The reference implementation (handle_stream.rs) handles one task per stream.
@@ -1162,7 +1162,7 @@ pub async fn start_tuic_server(
let client_proxy_selector = client_proxy_selector.clone();
let uuid = uuid.clone();
let password = password.clone();
let join_handle = tokio::spawn(async move {
let join_handle = crate::network_tasks::spawn(async move {
let mut tasks = crate::listener_tasks::ListenerTasks::immediate();
loop {
let conn = tokio::select! {
+4 -4
View File
@@ -189,7 +189,7 @@ async fn run_tun_server_inner(
let proxy_selector = proxy_selector.clone();
let resolver = resolver.clone();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
info!("Starting TCP connection handler");
let mut connections = JoinSet::new();
loop {
@@ -203,7 +203,7 @@ async fn run_tun_server_inner(
let proxy_selector = proxy_selector.clone();
let resolver = resolver.clone();
connections.spawn(async move {
crate::network_tasks::spawn_in(&mut connections, async move {
let remote_addr = new_conn.remote_addr;
let target = socket_addr_to_net_location(remote_addr);
@@ -233,7 +233,7 @@ async fn run_tun_server_inner(
let limits = config.resource_limits.clone();
let wake = tcp_stack.wake_handle();
tasks.spawn(async move {
crate::network_tasks::spawn_in(&mut tasks, async move {
handle_udp_packets(
udp_from_stack_rx,
udp_to_stack_tx,
@@ -417,7 +417,7 @@ async fn start_tun(
let (shutdown_tx, shutdown_rx) = oneshot::channel();
let (ready_tx, ready_rx) = oneshot::channel();
let handle = tokio::spawn(async move {
let handle = crate::network_tasks::spawn(async move {
let _keep_alive = shutdown_tx;
let mut ready = Some(ready_tx);
let result = run_tun_from_config_inner(
+2 -2
View File
@@ -262,7 +262,7 @@ impl TunUdpManager {
let (tx, rx) = mpsc::channel(CHANNEL_SIZE);
let handle = tokio::spawn(session_task(
let handle = crate::network_tasks::spawn(session_task(
peer_addr,
rx,
self.response_tx.clone(),
@@ -389,7 +389,7 @@ async fn session_task(
};
let (write_tx, write_rx) = mpsc::channel(CHANNEL_SIZE);
let handle = tokio::spawn(destination_task(
let handle = crate::network_tasks::spawn(destination_task(
peer_addr,
source_addr,
stream,
+5 -2
View File
@@ -111,11 +111,14 @@ pub(super) fn start(
incoming: incoming.clone(),
drops: drops.clone(),
};
workers.spawn(crate::control::url_test::BACKGROUND.scope((), worker.run()));
crate::network_tasks::spawn_in(
&mut workers,
crate::control::url_test::BACKGROUND.scope((), worker.run()),
);
}
drop(incoming);
let shutdown = cancel.clone();
tokio::spawn(async move {
crate::network_tasks::spawn(async move {
tokio::select! {
biased;
_ = shutdown.cancelled() => {},
+27 -6
View File
@@ -334,6 +334,22 @@ impl Prepared {
}
Ok(())
}
pub(crate) async fn refresh_peers(&mut self) -> stdio::Result<()> {
let mut failure = None;
for endpoint in &mut self.endpoints {
for (peer, current) in endpoint.config.peers.iter().zip(&mut endpoint.peers) {
match PreparedPeer::resolve(&endpoint.config, peer, &endpoint.outer_resolver).await
{
Ok(refreshed) => *current = refreshed,
Err(error) => {
failure.get_or_insert(error);
}
}
}
}
failure.map_or(Ok(()), Err)
}
}
struct EndpointRuntime {
@@ -351,10 +367,14 @@ pub(crate) struct Running {
endpoints: Vec<EndpointRuntime>,
}
impl Running {
pub async fn stop(&mut self) {
pub(crate) fn cancel(&self) {
for endpoint in &self.endpoints {
endpoint.cancel.cancel();
}
}
pub async fn stop(&mut self) {
self.cancel();
for mut endpoint in self.endpoints.drain(..) {
let _ = (&mut endpoint.task).await;
}
@@ -482,7 +502,7 @@ impl PreparedEndpoint {
let name = config.id.clone();
let cancel = CancellationToken::new();
let shutdown = cancel.clone();
let task = tokio::spawn(async move {
let task = tokio::spawn(crate::network_tasks::inherit(async move {
let mut tasks = JoinSet::new();
let mut inbound_udp_count = None;
if let Some((selector, resolver)) = inbound {
@@ -494,6 +514,7 @@ impl PreparedEndpoint {
report.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let failed = loop {
tokio::select! {
biased;
_ = shutdown.cancelled() => break false,
_ = device.wait() => break true,
_ = &mut stack.done => break true,
@@ -531,7 +552,7 @@ impl PreparedEndpoint {
stack.drops.load(Ordering::Relaxed)
);
drop(stack);
});
}));
Ok(EndpointRuntime { cancel, task })
}
}
@@ -558,13 +579,13 @@ fn spawn_inbound(
limits,
);
let udp_count = manager.destination_counter();
tasks.spawn(async move {
crate::network_tasks::spawn_in(tasks, async move {
if let Err(error) = manager.run().await {
log::warn!("WireGuard UDP forwarding stopped: {error}");
}
});
let mut tcp = stack.tcp.take().unwrap();
tasks.spawn(async move {
crate::network_tasks::spawn_in(tasks, async move {
let mut flows = JoinSet::new();
loop {
let connection = tokio::select! {
@@ -576,7 +597,7 @@ fn spawn_inbound(
};
let selector = selector.clone();
let resolver = resolver.clone();
flows.spawn(async move {
crate::network_tasks::spawn_in(&mut flows, async move {
log::debug!(
"WireGuard TCP {} -> {}",
connection.source,
+36
View File
@@ -106,6 +106,42 @@ async fn start_prepared(prepared: &Prepared) -> Running {
running
}
#[tokio::test]
async fn reconnect_resolves_peers_again_and_retries_without_replacing_connectors() {
#[derive(Debug)]
struct Answer(Option<SocketAddr>);
impl Resolver for Answer {
fn resolve_location(&self, _: &NetLocation) -> crate::resolver::ResolveFuture {
let address = self.0;
Box::pin(async move {
address.map(|address| vec![address]).ok_or_else(|| {
stdio::Error::new(stdio::ErrorKind::NotConnected, "network unavailable")
})
})
}
}
let mut configs = validate(&pair_config(51820)).unwrap();
let mut prepared = Prepared::new(&mut configs).await.unwrap();
let connector = prepared.endpoints[CLIENT].connector.clone();
let original = prepared.endpoints[CLIENT].peers[0].endpoints.clone();
prepared.endpoints[CLIENT].config.peers[0].endpoint = Some("peer.test:51820".into());
prepared.endpoints[CLIENT].outer_resolver = Arc::new(Answer(None));
let later = "192.0.2.10:51820".parse().unwrap();
prepared.endpoints[SERVER].config.peers[0].endpoint = Some("later.test:51820".into());
prepared.endpoints[SERVER].outer_resolver = Arc::new(Answer(Some(later)));
assert!(prepared.refresh_peers().await.is_err());
assert_eq!(prepared.endpoints[CLIENT].peers[0].endpoints, original);
assert_eq!(prepared.endpoints[SERVER].peers[0].endpoints, [later]);
let replacement = "192.0.2.9:51820".parse().unwrap();
prepared.endpoints[CLIENT].outer_resolver = Arc::new(Answer(Some(replacement)));
prepared.refresh_peers().await.unwrap();
assert_eq!(prepared.endpoints[CLIENT].peers[0].endpoints, [replacement]);
assert!(Arc::ptr_eq(
&connector,
&prepared.endpoints[CLIENT].connector
));
}
fn dns_answer(bytes: &[u8]) -> Vec<u8> {
use hickory_resolver::proto::{
op::{Message, MessageType},
+2 -2
View File
@@ -337,7 +337,7 @@ impl XhttpServer {
continue;
};
let state = self.state.clone();
requests.spawn(async move {
crate::network_tasks::spawn_in(&mut requests, async move {
let _request_permit = request_permit;
let Ok(Ok((mut request, stream))) = tokio::time::timeout(
Duration::from_secs(state.config.limits.setup_timeout_secs),
@@ -1262,7 +1262,7 @@ impl State {
continue;
};
let state = self.clone();
requests.spawn(async move {
crate::network_tasks::spawn_in(&mut requests, async move {
let _request_permit = request_permit;
let (parts, body) = request.into_parts();
let mut request = Request::from_parts(parts, ());
+5 -3
View File
@@ -535,10 +535,12 @@ impl Tasks {
tokio::select! { biased; _ = cancel.cancelled() => {}, _ = future => {} }
};
if crate::control::url_test::is_background() {
self.tracker
.spawn(crate::control::url_test::BACKGROUND.scope((), task));
crate::network_tasks::spawn(
self.tracker
.track_future(crate::control::url_test::BACKGROUND.scope((), task)),
);
} else {
self.tracker.spawn(task);
crate::network_tasks::spawn(self.tracker.track_future(task));
}
}
pub async fn wait(&self) {
+47
View File
@@ -10,6 +10,53 @@ use crate::{
tcp::{proxy_connector_impl::ProxyConnectorImpl, socket_connector_impl::SocketConnectorImpl},
};
#[tokio::test]
async fn network_reset_retires_h2_and_h3_request_workers() {
tokio::time::timeout(Duration::from_secs(10), async {
let certificate = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
let resolver: Arc<dyn Resolver> = Arc::new(NativeResolver::new());
for version in ["h2", "h3"] {
let target = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = reservation.local_addr().unwrap();
let protocol = json!({"type":"xhttp", "path":"/tunnel", "http_version":version,
"protocol":{"type":"vless","user_id":"550e8400-e29b-41d4-a716-446655440000"}});
let mut client_protocol = protocol.clone();
client_protocol["mode"] = json!("stream-one");
let (server, client) = if version == "h3" {
(json!({"address":address.to_string(), "transport":"quic",
"quic_settings":{"cert":certificate.cert.pem(),"key":certificate.signing_key.serialize_pem()}, "protocol":protocol}),
json!({"address":address.to_string(), "transport":"quic",
"quic_settings":{"verify":false,"sni_hostname":"localhost"},"protocol":client_protocol}))
} else {
(json!({"address":address.to_string(), "protocol":protocol}),
json!({"address":address.to_string(), "protocol":client_protocol}))
};
let server = crate::config::create_server_configs(serde_json::from_value(json!([server])).unwrap()).unwrap().configs.remove(0);
let network = crate::network_tasks::NetworkLifetime::default();
drop(reservation);
let handles = network.scope(crate::tcp::tcp_server::start_servers(server, resolver.clone())).await.unwrap();
let config: crate::config::ResolvedClientConfig = crate::config::test_support::from_json(client).unwrap();
let socket = SocketConnectorImpl::from_config(&config, Some(&config.address)).unwrap();
let proxy = ProxyConnectorImpl::from_config(config, resolver.clone()).unwrap().unwrap();
let chain = ClientProxyChain::new(vec![InitialHopEntry::Proxy {
socket: Arc::new(socket), proxy: Box::new(proxy),
}], vec![]);
let destination = crate::address::NetLocation::from_str(&target.local_addr().unwrap().to_string(), None).unwrap();
let mut stream = chain.connect_tcp(destination.into(), &resolver).await.unwrap().client_stream;
stream.write_all(b"x").await.unwrap();
stream.flush().await.unwrap();
let (mut accepted, _) = target.accept().await.unwrap();
assert_eq!(accepted.read_u8().await.unwrap(), b'x');
network.cancel();
network.wait().await;
let result = accepted.read(&mut [0]).await;
assert!(matches!(result, Ok(0)) || result.is_err(), "{result:?}");
for handle in handles { let _ = handle.await; }
}
}).await.unwrap();
}
#[tokio::test]
async fn all_modes_and_carriers_exchange_data_before_upload_eof() {
let certificate = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();