diff --git a/easytier/src/common/netns.rs b/easytier/src/common/netns.rs index a998e386..dde9d583 100644 --- a/easytier/src/common/netns.rs +++ b/easytier/src/common/netns.rs @@ -18,7 +18,7 @@ impl NetNSGuard { pub fn new(ns: Option) -> Box { let old_ns = if ns.is_some() { let old_ns = if cfg!(target_os = "linux") { - Some(std::fs::File::open("/proc/self/ns/net").unwrap()) + Some(std::fs::File::open("/proc/thread-self/ns/net").unwrap()) } else { None }; @@ -101,8 +101,13 @@ impl NetNS { { // TODO: do we really need this lock // let _lock = LOCK.lock().await; - let _guard = NetNSGuard::new(self.name.clone()); - f().await + // Namespace membership belongs to a thread. Restore it before yielding, + // and re-enter on every poll even if the future moves to another thread. + // Drop the future in its namespace too, including on cancellation. + let mut future = guarden::guard::ContextGuard::new(Box::pin(self.run(f)), |future| { + self.run(|| drop(future)); + }); + std::future::poll_fn(|cx| self.run(|| future.as_mut().poll(cx))).await } pub fn run(&self, f: F) -> Ret diff --git a/easytier/src/tests/network.rs b/easytier/src/tests/network.rs index cf27fc29..bf0eed54 100644 --- a/easytier/src/tests/network.rs +++ b/easytier/src/tests/network.rs @@ -93,3 +93,58 @@ impl Drop for TestNetwork { } } } + +#[test] +fn nested_namespace_guard_restores_current_thread() { + let mut network = TestNetwork::new(); + network.create_empty_namespace("net_a"); + network.create_empty_namespace("net_b"); + let net_a = NetNS::new(Some(network.namespace("net_a").into())); + let net_b = NetNS::new(Some(network.namespace("net_b").into())); + let current = || std::fs::read_link("/proc/thread-self/ns/net").unwrap(); + let original = current(); + + net_a.run(|| { + let expected = current(); + assert_ne!(expected, original); + net_b.run(|| assert_ne!(current(), expected)); + assert_eq!(current(), expected); + }); + assert_eq!(current(), original); +} + +#[tokio::test] +async fn async_namespace_restores_between_polls() { + let mut network = TestNetwork::new(); + network.create_empty_namespace("net_a"); + network.create_empty_namespace("net_b"); + let net_a = NetNS::new(Some(network.namespace("net_a").into())); + let net_b = NetNS::new(Some(network.namespace("net_b").into())); + let current = || std::fs::read_link("/proc/thread-self/ns/net").unwrap(); + let original = current(); + let expected = net_a.run(current); + + let mut future = Box::pin(net_a.run_async(|| async { + assert_eq!(current(), expected); + tokio::task::yield_now().await; + assert_eq!(current(), expected); + })); + assert!(futures::poll!(future.as_mut()).is_pending()); + assert_eq!(current(), original); + net_b.run(|| assert_ne!(current(), expected)); + future.await; + assert_eq!(current(), original); + + let dropped_in = std::cell::RefCell::new(None); + let mut cancelled = Box::pin(net_a.run_async(|| async { + let _cleanup = guarden::guard::ContextGuard::new((), |_| { + *dropped_in.borrow_mut() = Some(current()); + }); + std::future::pending::<()>().await; + })); + assert!(futures::poll!(cancelled.as_mut()).is_pending()); + assert_eq!(current(), original); + drop(cancelled); + assert_eq!(current(), original); + assert_eq!(*dropped_in.borrow(), Some(expected)); +} diff --git a/easytier/src/tests/three_node.rs b/easytier/src/tests/three_node.rs index 5d07ba06..e1810d20 100644 --- a/easytier/src/tests/three_node.rs +++ b/easytier/src/tests/three_node.rs @@ -2836,13 +2836,16 @@ pub async fn port_forward_test( let mut buf = vec![0; buf_size as usize]; rand::thread_rng().fill(&mut buf[..]); + // Allow the UDP connector's three-second handshake deadline to complete. + // Echo and shutdown retain their separate one-second deadlines. + let connect_timeout = Duration::from_secs(5); _tunnel_pingpong_netns_with_timeout( tcp_listener, tcp_connector, NetNS::new(Some(network.namespace("net_c").into())), NetNS::new(Some(network.namespace("net_a").into())), buf, - Duration::from_secs(1), + connect_timeout, ) .await .unwrap(); @@ -2859,7 +2862,7 @@ pub async fn port_forward_test( NetNS::new(Some(network.namespace("net_d").into())), NetNS::new(Some(network.namespace("net_a").into())), buf, - Duration::from_secs(1), + connect_timeout, ) .await .unwrap(); @@ -2876,7 +2879,7 @@ pub async fn port_forward_test( NetNS::new(Some(network.namespace("net_c").into())), NetNS::new(Some(network.namespace("net_a").into())), buf, - Duration::from_secs(1), + connect_timeout, ) .await .unwrap(); @@ -2893,7 +2896,7 @@ pub async fn port_forward_test( NetNS::new(Some(network.namespace("net_d").into())), NetNS::new(Some(network.namespace("net_a").into())), buf, - Duration::from_secs(1), + connect_timeout, ) .await .unwrap(); diff --git a/easytier/src/tunnel/common.rs b/easytier/src/tunnel/common.rs index 71e8f447..908d2ab0 100644 --- a/easytier/src/tunnel/common.rs +++ b/easytier/src/tunnel/common.rs @@ -252,8 +252,10 @@ pub(crate) mod tests { use atomic_shim::AtomicU64; use std::{sync::Arc, time::Instant}; + use anyhow::Context as _; use futures::{Future, SinkExt, StreamExt}; use tokio_util::bytes::{BufMut, Bytes, BytesMut}; + use tokio_util::task::AbortOnDropHandle; use easytier_core::{ connectivity::protocol::raw::TunnelDialer, packet::ZCPacket, socket::SocketListener, @@ -341,7 +343,7 @@ pub(crate) mod tests { if !once { while let Some(item) = recv.next().await { let Ok(msg) = item else { - continue; + break; }; tracing::debug!(?msg, "recv a msg, try echo back"); if send.send(msg).await.is_err() { @@ -368,23 +370,41 @@ pub(crate) mod tests { tracing::warn!("echo server exit..."); } + #[tokio::test] + async fn echo_server_stops_on_read_error() { + use easytier_core::tunnel::{TunnelError, wrapper::TunnelWrapper}; + + let mut failed = false; + let reader = futures::stream::poll_fn(move |_| { + assert!(!failed, "a terminal read error must not be polled again"); + failed = true; + std::task::Poll::Ready(Some(Err::(TunnelError::IOError( + std::io::ErrorKind::NotConnected.into(), + )))) + }); + let writer = futures::sink::drain().sink_map_err(|err| -> TunnelError { match err {} }); + + _tunnel_echo_server(Box::new(TunnelWrapper::new(reader, writer, None)), false).await; + } + async fn _tunnel_pingpong_netns( mut listener: L, connector: C, l_netns: NetNS, c_netns: NetNS, buf: Vec, - ) where + timeout: std::time::Duration, + ) -> anyhow::Result<()> + where L: SocketListener> + Sync + 'static, C: TunnelDialer, { - l_netns - .run_async(|| async { - listener.listen().await.unwrap(); - }) - .await; + tokio::time::timeout(timeout, l_netns.run_async(|| listener.listen())) + .await + .context("listener setup timed out")? + .context("listener setup failed")?; - let lis = tokio::spawn(async move { + let lis = AbortOnDropHandle::new(tokio::spawn(async move { let ret = listener.accept().await.unwrap(); println!("accept: {:?}", ret.info()); assert_eq!( @@ -392,9 +412,12 @@ pub(crate) mod tests { listener.local_url() ); _tunnel_echo_server(ret, false).await - }); + })); - let tunnel = c_netns.run_async(|| connector.connect()).await.unwrap(); + let tunnel = tokio::time::timeout(timeout, c_netns.run_async(|| connector.connect())) + .await + .context("connect timed out")? + .context("connect failed")?; println!("connect: {:?}", tunnel.info()); if connector.remote_url().scheme() == "faketcp" { @@ -409,27 +432,37 @@ pub(crate) mod tests { let (mut recv, mut send) = tunnel.split(); - send.send(ZCPacket::new_with_payload(buf.as_slice())) - .await - .unwrap(); + tokio::time::timeout( + timeout, + send.send(ZCPacket::new_with_payload(buf.as_slice())), + ) + .await + .context("send timed out")? + .context("send failed")?; let ret = tokio::time::timeout(tokio::time::Duration::from_secs(1), recv.next()) .await - .unwrap() - .unwrap() - .unwrap(); + .context("echo timed out")? + .context("tunnel closed before echo")? + .context("echo read failed")?; println!("echo back: {:?}", ret); assert_eq!(ret.payload(), Bytes::from(buf)); - send.close().await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(1), send.close()) + .await + .context("client close timed out")? + .context("client close failed")?; if ["udp", "wg"].contains(&connector.remote_url().scheme()) { lis.abort(); + let _ = lis.await; } else { // lis should finish in 1 second let ret = tokio::time::timeout(tokio::time::Duration::from_secs(1), lis).await; - assert!(ret.is_ok()); + ret.context("echo server shutdown timed out")? + .context("echo server task failed")?; } + Ok(()) } pub(crate) async fn _tunnel_pingpong_netns_with_timeout( @@ -444,30 +477,31 @@ pub(crate) mod tests { L: SocketListener> + Sync + 'static, C: TunnelDialer, { - let handle = tokio::spawn(async move { - _tunnel_pingpong_netns(listener, connector, l_netns, c_netns, buf).await; - }); + // Each phase has its own deadline; connection setup must not consume + // the echo or shutdown budget. Dropping the caller cancels both tasks. + let handle = AbortOnDropHandle::new(tokio::spawn(async move { + _tunnel_pingpong_netns(listener, connector, l_netns, c_netns, buf, timeout).await + })); - match tokio::time::timeout(timeout, handle).await { - Ok(join_res) => match join_res { - Ok(_) => Ok(()), - Err(join_err) => { - if join_err.is_panic() { - let payload = join_err.into_panic(); - let msg = match payload.downcast::() { - Ok(s) => *s, - Err(payload) => match payload.downcast::<&str>() { - Ok(s) => (*s).to_string(), - Err(_) => "non-string panic payload".to_string(), - }, - }; - Err(anyhow::anyhow!("task panicked: {}", msg)) - } else { - Err(anyhow::anyhow!("task cancelled")) - } + match handle.await { + Ok(result) => { + result.inspect_err(|error| eprintln!("tunnel pingpong failed: {error:#}")) + } + Err(join_err) => { + if join_err.is_panic() { + let payload = join_err.into_panic(); + let msg = match payload.downcast::() { + Ok(s) => *s, + Err(payload) => match payload.downcast::<&str>() { + Ok(s) => (*s).to_string(), + Err(_) => "non-string panic payload".to_string(), + }, + }; + Err(anyhow::anyhow!("task panicked: {}", msg)) + } else { + Err(anyhow::anyhow!("task cancelled")) } - }, - Err(elapsed) => Err(elapsed.into()), + } } } diff --git a/easytier/src/tunnel/quic.rs b/easytier/src/tunnel/quic.rs index 339b82a2..d6861546 100644 --- a/easytier/src/tunnel/quic.rs +++ b/easytier/src/tunnel/quic.rs @@ -673,7 +673,7 @@ mod tests { socket::SocketListener, socket::udp::{ UdpBindOptions, UdpSessionAcceptKind, UdpSessionListenRequest, UdpSessionProtocol, - VirtualUdpSocket, + UdpSessionSocket, VirtualUdpSocket, }, }; use futures::{SinkExt, StreamExt}; @@ -692,7 +692,9 @@ mod tests { let mut listener = new_runtime_udp_session_listener( format!("quic://{bind_addr}").parse().unwrap(), UdpSessionListenRequest::new( - UdpBindOptions::port_bound_listener(bind_addr).with_only_v6(false), + // Port-bound listeners enable address reuse on Unix. This test + // needs an exclusive ephemeral port across parallel processes. + UdpBindOptions::direct_connect().with_local_addr(Some(bind_addr)), ), UdpSessionAcceptKind::Classified(UdpSessionProtocol::Quic), NetNS::new(None), @@ -716,7 +718,9 @@ mod tests { Endpoint::new_with_abstract_socket(endpoint_config(), None, socket, runtime) .unwrap(); endpoint.set_default_client_config(client_config()); + let client_addr = endpoint.local_addr().unwrap(); + let (first_accepted_tx, first_accepted_rx) = tokio::sync::oneshot::channel(); let server_task = tokio::spawn(async move { let session = listener.accept().await.unwrap(); let admission = ServerProtocolAdmissionController::new(1, 2) @@ -724,12 +728,17 @@ mod tests { .unwrap(); let mut accepted = QuicAcceptedSession::new(session, local_url, admission).unwrap(); let first = accepted.accept().await.unwrap(); + first_accepted_tx.send(()).unwrap(); let second = accepted.accept().await.unwrap(); + let extra_session = + tokio::time::timeout(Duration::from_millis(50), listener.accept()).await; assert!( - tokio::time::timeout(Duration::from_millis(50), listener.accept()) - .await - .is_err(), - "both QUIC connections must use the first accepted UDP session" + extra_session.is_err(), + "both QUIC connections must use the first accepted UDP session; \ + server={remote_addr}, client={client_addr}, extra_peer={:?}", + extra_session + .as_ref() + .map(|result| result.as_ref().map(|session| session.peer_addr())) ); (first, second) }); @@ -745,6 +754,9 @@ mod tests { .send(ZCPacket::new_with_payload(b"first QUIC connection")) .await .unwrap(); + // Keep both connections alive, but establish their server-side identity + // before starting the second handshake, which may complete out of order. + first_accepted_rx.await.unwrap(); let second_connection = endpoint .connect(remote_addr, "localhost") .unwrap()