From 0cfaa3c42f83949a8044b03f55064e25b3be64fb Mon Sep 17 00:00:00 2001 From: Philipp Krones Date: Fri, 5 Jun 2026 12:45:05 +0200 Subject: [PATCH 1/3] [memcached] Fix TCP address resolution PR #7112 added support for Unix sockets to the memcached backend. As a side effect, it switched the TCP address parsing from being done inside of `TcpStream::connect` to `SocketAddr::parse`. The major difference is that `SocketAddr::parse` cannot resolve addresses like `localhost:1234`, as it errors out if it sees non-octal numbers in the address. This seemed like an unintended breaking change to me. I also can't see the benefit of doing that, as `TcpStream::connect` already calls `to_socket_addrs` internally. --- core/services/memcached/src/core.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/core/services/memcached/src/core.rs b/core/services/memcached/src/core.rs index 1221bb08ed7f..2b9a49533245 100644 --- a/core/services/memcached/src/core.rs +++ b/core/services/memcached/src/core.rs @@ -23,7 +23,6 @@ use fastpool::bounded; use opendal_core::raw::*; use opendal_core::*; use std::io; -use std::net::SocketAddr; use std::pin::Pin; use std::task::{Context, Poll}; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; @@ -42,10 +41,7 @@ pub enum SocketStream { impl SocketStream { pub async fn connect_tcp(addr_str: &str) -> io::Result { - let socket_addr: SocketAddr = addr_str - .parse() - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - let stream = TcpStream::connect(socket_addr).await?; + let stream = TcpStream::connect(addr_str).await?; Ok(SocketStream::Tcp(stream)) } From dc56ffea62b0b35c875d8a28cd4d6fc0a8f26564 Mon Sep 17 00:00:00 2001 From: Philipp Krones Date: Mon, 22 Jun 2026 15:04:42 +0200 Subject: [PATCH 2/3] Add regression tests --- core/services/memcached/src/core.rs | 53 +++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/core/services/memcached/src/core.rs b/core/services/memcached/src/core.rs index 2b9a49533245..bfd7fa601ef3 100644 --- a/core/services/memcached/src/core.rs +++ b/core/services/memcached/src/core.rs @@ -219,3 +219,56 @@ impl MemcachedCore { conn.delete(&percent_encode_path(key)).await } } + +#[cfg(test)] +mod tests { + use super::*; + use tokio::net::TcpListener; + + #[tokio::test] + async fn connect_tcp_socket() -> std::io::Result<()> { + // Test both ip and localhost addresses to make sure that both can be parsed. + for addr in &["127.0.0.1:11211", "localhost:11211", "[::1]:11211"] { + let listener = TcpListener::bind(addr).await?; + let addr = listener.local_addr()?.to_string(); + + let accepted = tokio::spawn(async move { + let _ = listener.accept().await?; + Ok::<_, std::io::Error>(()) + }); + + let _stream = SocketStream::connect_tcp(&addr).await?; + accepted.await.unwrap()?; + } + + Ok(()) + } + + #[cfg(unix)] + #[tokio::test] + async fn connect_unix_socket() -> std::io::Result<()> { + use std::time::{SystemTime, UNIX_EPOCH}; + use tokio::net::UnixListener; + + let path = std::env::temp_dir().join(format!( + "opendal-memcached-{}.sock", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + + let listener = UnixListener::bind(&path)?; + + let accepted = tokio::spawn(async move { + let _ = listener.accept().await?; + Ok::<_, std::io::Error>(()) + }); + + let _stream = SocketStream::connect_unix(path.to_str().unwrap()).await?; + accepted.await.unwrap()?; + + let _ = std::fs::remove_file(path); + Ok(()) + } +} From 521a915f2b940f5a25e43ba72906addd3f638c38 Mon Sep 17 00:00:00 2001 From: Philipp Krones Date: Mon, 22 Jun 2026 17:31:03 +0200 Subject: [PATCH 3/3] Update test comments --- core/services/memcached/src/core.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/core/services/memcached/src/core.rs b/core/services/memcached/src/core.rs index bfd7fa601ef3..a272a2b5186f 100644 --- a/core/services/memcached/src/core.rs +++ b/core/services/memcached/src/core.rs @@ -225,9 +225,11 @@ mod tests { use super::*; use tokio::net::TcpListener; + // regression test for connecting to a webdav server. + // Because setting up a dedicated behavior test is expensive, we choose to test `SocketStream::connect_tcp` instead. + // In the future, we could set up a webdav server to test TCP connection properly. #[tokio::test] async fn connect_tcp_socket() -> std::io::Result<()> { - // Test both ip and localhost addresses to make sure that both can be parsed. for addr in &["127.0.0.1:11211", "localhost:11211", "[::1]:11211"] { let listener = TcpListener::bind(addr).await?; let addr = listener.local_addr()?.to_string(); @@ -244,6 +246,9 @@ mod tests { Ok(()) } + // regression test for connecting to a webdav server. + // Because setting up a dedicated behavior test is expensive, we choose to test `SocketStream::connect_unix` instead. + // In the future, we could set up a webdav server to test UNIX socket connection properly. #[cfg(unix)] #[tokio::test] async fn connect_unix_socket() -> std::io::Result<()> {