laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-10-05 19:18:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tcp): try remaining DNS addresses after connection failure

Browse Source 
TCP connections failed when the first resolved address was unreachable even if another address worked. Retain all resolver candidates for both connection paths and let Tokio try them within one connection timeout.

Cover fallback with a real listener and exercise hostname connections through connect-on-send and background connect.
This commit is contained in:
Martti Malmi
2026-09-30 17:14:50 +00:00
committed by Johnathan Corgan
parent 1b55a0def3
commit 712faebddd
3 changed files with 153 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+2
View File
@@ -41,6 +41,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
reopened after the interface goes away. A running node whose interface was
removed and not recreated retried the reopen on every beacon interval and
leaked one descriptor each time.
- TCP connections try the remaining addresses for a hostname after a
connection fails, within the existing overall connection timeout.
## [0.5.2] - 2026-09-28
src/transport/mod.rs
+13 -18
View File
@@ -1082,28 +1082,23 @@ impl TransportHandle {
pub(crate) async fn resolve_socket_addr(
addr: &TransportAddr,
) -> Result<SocketAddr, TransportError> {
resolve_socket_addrs(addr).await?.next().ok_or_else(|| {
TransportError::InvalidAddress(format!("DNS resolution returned no addresses for {}", addr))
})
}
/// Resolve every socket address in resolver order, bypassing DNS for numeric IPs.
pub(crate) async fn resolve_socket_addrs(
addr: &TransportAddr,
) -> Result<impl Iterator<Item = SocketAddr>, TransportError> {
let s = addr
.as_str()
.ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
// Fast path: numeric IP address — no DNS lookup
if let Ok(sock_addr) = s.parse::<SocketAddr>() {
return Ok(sock_addr);
}
// Slow path: DNS resolution
tokio::net::lookup_host(s)
.await
.map_err(|e| {
TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
})?
.next()
.ok_or_else(|| {
TransportError::InvalidAddress(format!(
"DNS resolution returned no addresses for {}",
s
))
})
// lookup_host handles numeric addresses without allocating or querying DNS.
tokio::net::lookup_host(s).await.map_err(|e| {
TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
})
}
// ============================================================================
src/transport/tcp/mod.rs
+138 -50
View File
@@ -25,7 +25,7 @@
mod pool;
pub mod stats;
use super::resolve_socket_addr;
use super::resolve_socket_addrs;
use super::{
ConnectionState, DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr,
TransportError, TransportId, TransportState, TransportType,
@@ -379,25 +379,20 @@ impl TcpTransport {
&self,
addr: &TransportAddr,
) -> Result<Arc<Mutex<OwnedWriteHalf>>, TransportError> {
let socket_addr = resolve_socket_addr(addr).await?;
let socket_addrs: Vec<_> = resolve_socket_addrs(addr).await?.collect();
let timeout_ms = self.config.connect_timeout_ms();
// Connect with timeout
let stream = match tokio::time::timeout(
Duration::from_millis(timeout_ms),
TcpStream::connect(socket_addr),
)
.await
{
Ok(Ok(stream)) => stream,
Ok(Err(_)) => {
let stream = match connect_to_any_addr(&socket_addrs, timeout_ms).await {
Ok(stream) => stream,
Err(error @ TransportError::ConnectionRefused) => {
self.stats.record_connect_refused();
return Err(TransportError::ConnectionRefused);
return Err(error);
}
Err(_) => {
Err(error @ TransportError::Timeout) => {
self.stats.record_connect_timeout();
return Err(TransportError::Timeout);
return Err(error);
}
Err(error) => return Err(error),
};
// Configure socket options via socket2
@@ -518,10 +513,8 @@ impl TcpTransport {
}
// Validate address is UTF-8 before spawning (fail fast on bad input)
let addr_string = addr
.as_str()
.ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?
.to_string();
addr.as_str()
.ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
let timeout_ms = self.config.connect_timeout_ms();
let config = self.config.clone();
let transport_id = self.transport_id;
@@ -536,51 +529,28 @@ impl TcpTransport {
let task = tokio::spawn(async move {
// Resolve address (may involve DNS for hostnames)
let socket_addr: SocketAddr = if let Ok(sa) = addr_string.parse() {
sa
} else {
tokio::net::lookup_host(&addr_string)
.await
.map_err(|e| {
TransportError::InvalidAddress(format!(
"DNS resolution failed for {}: {}",
addr_string, e
))
})?
.next()
.ok_or_else(|| {
TransportError::InvalidAddress(format!(
"DNS resolution returned no addresses for {}",
addr_string
))
})?
};
let socket_addrs: Vec<_> = resolve_socket_addrs(&remote_addr).await?.collect();
// Connect with timeout
let stream = match tokio::time::timeout(
Duration::from_millis(timeout_ms),
TcpStream::connect(socket_addr),
)
.await
{
Ok(Ok(stream)) => stream,
Ok(Err(e)) => {
let stream = match connect_to_any_addr(&socket_addrs, timeout_ms).await {
Ok(stream) => stream,
Err(error @ TransportError::ConnectionRefused) => {
debug!(
transport_id = %transport_id,
remote_addr = %remote_addr,
error = %e,
error = %error,
"Background TCP connect refused"
);
return Err(TransportError::ConnectionRefused);
return Err(error);
}
Err(_) => {
Err(error @ TransportError::Timeout) => {
debug!(
transport_id = %transport_id,
remote_addr = %remote_addr,
"Background TCP connect timed out"
);
return Err(TransportError::Timeout);
return Err(error);
}
Err(error) => return Err(error),
};
// Configure socket options via socket2
@@ -1123,6 +1093,32 @@ async fn tcp_receive_loop(
// Socket Configuration Helpers
// ============================================================================
async fn connect_to_any_addr(
socket_addrs: &[SocketAddr],
timeout_ms: u64,
) -> Result<TcpStream, TransportError> {
if socket_addrs.is_empty() {
return Err(TransportError::InvalidAddress(
"DNS resolution returned no addresses".into(),
));
}
// Try candidates in resolver order within one overall connection timeout.
match tokio::time::timeout(
Duration::from_millis(timeout_ms),
TcpStream::connect(socket_addrs),
)
.await
{
Ok(Ok(stream)) => Ok(stream),
Ok(Err(error)) => {
trace!(error = %error, "TCP connection candidates failed");
Err(TransportError::ConnectionRefused)
}
Err(_) => Err(TransportError::Timeout),
}
}
/// Configure a TCP socket with the transport's settings.
fn configure_socket(
stream: &std::net::TcpStream,
@@ -1264,6 +1260,98 @@ mod tests {
}
}
#[tokio::test]
async fn test_connect_tries_later_candidates() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let good_addr = listener.local_addr().unwrap();
// No listener can bind port zero: binding it allocates an ephemeral port.
let bad_addr = "127.0.0.1:0".parse().unwrap();
let stream = connect_to_any_addr(&[bad_addr, good_addr], 1_000)
.await
.expect("second TCP candidate should connect");
let (accepted, _) = timeout(Duration::from_secs(1), listener.accept())
.await
.unwrap()
.unwrap();
assert_eq!(stream.peer_addr().unwrap(), good_addr);
assert_eq!(accepted.peer_addr().unwrap(), stream.local_addr().unwrap());
}
#[tokio::test]
async fn test_connect_candidates_fail() {
assert!(matches!(
connect_to_any_addr(&[], 1_000).await,
Err(TransportError::InvalidAddress(_))
));
assert!(matches!(
connect_to_any_addr(&["127.0.0.1:0".parse().unwrap()], 1_000).await,
Err(TransportError::ConnectionRefused)
));
}
async fn send_to_hostname(background: bool) {
// Prefer the last localhost address so dual-stack hosts also
// exercise fallback through the full transport connection path.
let mut addresses: Vec<_> = tokio::net::lookup_host("localhost:0")
.await
.unwrap()
.collect();
addresses.reverse();
let listener = TcpListener::bind(addresses.as_slice()).await.unwrap();
let (tx, _rx) = packet_channel(10);
let config = make_outbound_config();
// A refused localhost connection can take over two seconds on Windows.
// Allow the configured connection timeout, plus DNS/scheduling margin.
let connect_deadline =
Duration::from_millis(config.connect_timeout_ms()) + Duration::from_secs(2);
let mut sender = TcpTransport::new(TransportId::new(1), None, config, tx);
sender.start_async().await.unwrap();
let remote = TransportAddr::from_string(&format!(
"localhost:{}",
listener.local_addr().unwrap().port()
));
if background {
sender.connect_async(&remote).await.unwrap();
assert!(
wait_until(
|| sender.connection_state_sync(&remote) == ConnectionState::Connected,
connect_deadline,
)
.await
);
}
timeout(
connect_deadline,
sender.send_async(&remote, &build_msg1_frame()),
)
.await
.expect("hostname send exceeded connection deadline")
.unwrap();
let (mut stream, _) = timeout(Duration::from_secs(2), listener.accept())
.await
.unwrap()
.unwrap();
let packet = timeout(Duration::from_secs(2), read_fmp_packet(&mut stream, 1400))
.await
.unwrap()
.unwrap();
assert_eq!(packet, build_msg1_frame());
sender.stop_async().await.unwrap();
}
#[tokio::test]
async fn test_connect_on_send_hostname() {
send_to_hostname(false).await;
}
#[tokio::test]
async fn test_connect_async_hostname() {
send_to_hostname(true).await;
}
/// Listener on every local address, so that two connections can reach
/// it from one source port on two different local addresses.
fn wildcard_config() -> TcpConfig {