more lib fix

lib/vey-http/src/body/chunked_decoder.rs

@@ -174,8 +174,15 @@ impl ChunkedDataDecodeReaderInternal {
 
                     match memchr::memchr(b'\n', r_buf) {
                         Some(p) => {
-                            self.chunk_header.put_slice(&r_buf[0..=p]);
-                            reader.as_mut().consume(p + 1);
+                            let len = p + 1;
+                            if self.chunk_header.len() + len > self.body_line_max_size {
+                                return Poll::Ready(Err(io::Error::other(format!(
+                                    "chunk header line too long (> {})",
+                                    self.body_line_max_size
+                                ))));
+                            }
+                            self.chunk_header.put_slice(&r_buf[..len]);
+                            reader.as_mut().consume(len);
                             break;
                         }
                         None => {
@@ -342,4 +349,49 @@ mod test {
         assert_eq!(&buf[..len], b"body");
         assert!(decoder.finished());
     }
+
+    #[tokio::test]
+    async fn chunk_line_in_one_read_respects_max_size() {
+        let content = b"123456789\n0\r\n\r\n";
+        let stream = tokio_test::io::Builder::new().read(content).build();
+        let mut buf_stream = BufReader::new(stream);
+        let mut decoder = ChunkedDataDecodeReader::new(&mut buf_stream, 8);
+
+        let mut buf = [0u8; 16];
+        let err = decoder.read(&mut buf).await.unwrap_err();
+        assert!(
+            err.to_string().contains("chunk header line too long"),
+            "{err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn chunk_line_split_across_reads_respects_max_size() {
+        let stream = tokio_test::io::Builder::new()
+            .read(b"1234")
+            .read(b"56789\n0\r\n\r\n")
+            .build();
+        let mut buf_stream = BufReader::new(stream);
+        let mut decoder = ChunkedDataDecodeReader::new(&mut buf_stream, 8);
+
+        let mut buf = [0u8; 16];
+        let err = decoder.read(&mut buf).await.unwrap_err();
+        assert!(
+            err.to_string().contains("chunk header line too long"),
+            "{err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn chunk_line_at_max_size_is_accepted() {
+        let content = b"0\r\n\r\n";
+        let stream = tokio_test::io::Builder::new().read(content).build();
+        let mut buf_stream = BufReader::new(stream);
+        let mut decoder = ChunkedDataDecodeReader::new(&mut buf_stream, 3);
+
+        let mut buf = [0u8; 8];
+        let len = decoder.read(&mut buf).await.unwrap();
+        assert_eq!(len, 0);
+        assert!(decoder.finished());
+    }
 }

lib/vey-redis-client/Cargo.toml

@@ -9,7 +9,7 @@ description = "Redis client for VEY services"
 [dependencies]
 anyhow.workspace = true
 redis = { workspace = true, features = ["aio", "tokio-comp"] }
-tokio = { workspace = true, features = ["net"] }
+tokio = { workspace = true, features = ["macros", "net", "rt", "time"] }
 tokio-rustls.workspace = true
 rustls-pki-types.workspace = true
 yaml-rust = { workspace = true, optional = true }

lib/vey-redis-client/src/lib.rs

@@ -12,6 +12,7 @@ use redis::aio::MultiplexedConnection;
 use redis::{AsyncCommands, AsyncConnectionConfig, ProtocolVersion, RedisConnectionInfo};
 use rustls_pki_types::ServerName;
 use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpStream;
 use tokio_rustls::TlsConnector;
 
 use vey_types::net::{Host, RustlsClientConfig, RustlsClientConfigBuilder, UpstreamAddr};
@@ -138,21 +139,32 @@ impl RedisClientConfigBuilder {
 }
 
 impl RedisClientConfig {
-    async fn lookup_server(&self) -> anyhow::Result<SocketAddr> {
+    async fn lookup_server(&self) -> anyhow::Result<Vec<SocketAddr>> {
         match self.server.host() {
             Host::Domain(domain) => {
-                let mut ips = tokio::net::lookup_host((domain.as_str(), self.server.port()))
-                    .await
-                    .map_err(|e| anyhow!("failed to resolve domain {domain}: {e}"))?;
-                ips.next()
-                    .ok_or_else(|| anyhow!("no ip address resolved for domain {domain}"))
+                let ips = match tokio::time::timeout(
+                    self.connect_timeout,
+                    tokio::net::lookup_host((domain.as_str(), self.server.port())),
+                )
+                .await
+                {
+                    Ok(Ok(ips)) => ips,
+                    Ok(Err(e)) => {
+                        return Err(anyhow!("failed to resolve domain {domain}: {e}"));
+                    }
+                    Err(_) => return Err(anyhow!("timeout to resolve domain {domain}")),
+                };
+                let addrs: Vec<SocketAddr> = ips.collect();
+                if addrs.is_empty() {
+                    return Err(anyhow!("no ip address resolved for domain {domain}"));
+                }
+                Ok(addrs)
             }
-            Host::Ip(ip) => Ok(SocketAddr::new(*ip, self.server.port())),
+            Host::Ip(ip) => Ok(vec![SocketAddr::new(*ip, self.server.port())]),
         }
     }
 
-    pub async fn connect(&self) -> anyhow::Result<impl AsyncCommands + use<>> {
-        let peer = self.lookup_server().await?;
+    async fn tcp_connect(&self, peer: SocketAddr) -> anyhow::Result<TcpStream> {
         let socket = vey_socket::tcp::new_socket_to(
             peer.ip(),
             &Default::default(),
@@ -162,27 +174,47 @@ impl RedisClientConfig {
         )
         .map_err(|e| anyhow!("failed to create new socket: {e}"))?;
 
-        let stream = match tokio::time::timeout(self.connect_timeout, socket.connect(peer)).await {
-            Ok(Ok(stream)) => stream,
-            Ok(Err(e)) => return Err(anyhow!("failed to connect to {}: {e}", self.server)),
-            Err(_) => return Err(anyhow!("timeout to connect to {}", self.server)),
-        };
+        match tokio::time::timeout(self.connect_timeout, socket.connect(peer)).await {
+            Ok(Ok(stream)) => Ok(stream),
+            Ok(Err(e)) => Err(anyhow!("failed to connect to {}: {e}", self.server)),
+            Err(_) => Err(anyhow!("timeout to connect to {}", self.server)),
+        }
+    }
 
-        if let Some(tls_client) = &self.tls_client {
-            let tls_connector = TlsConnector::from(tls_client.tls_config.driver.clone());
-            match tokio::time::timeout(
-                tls_client.tls_config.handshake_timeout,
-                tls_connector.connect(tls_client.tls_name.clone(), stream),
-            )
-            .await
-            {
-                Ok(Ok(stream)) => self.redis_handshake(stream).await,
-                Ok(Err(e)) => Err(anyhow!("failed to tls handshake with {}: {e}", self.server)),
-                Err(_) => Err(anyhow!("timeout to tls handshake with {}", self.server)),
-            }
-        } else {
-            self.redis_handshake(stream).await
+    pub async fn connect(&self) -> anyhow::Result<impl AsyncCommands + use<>> {
+        let peers = self.lookup_server().await?;
+        self.connect_peers(peers).await
+    }
+
+    async fn connect_peers(&self, peers: Vec<SocketAddr>) -> anyhow::Result<MultiplexedConnection> {
+        let mut last_err = None;
+        for peer in peers {
+            let stream = match self.tcp_connect(peer).await {
+                Ok(stream) => stream,
+                Err(e) => {
+                    last_err = Some(e);
+                    continue;
+                }
+            };
+
+            return if let Some(tls_client) = &self.tls_client {
+                let tls_connector = TlsConnector::from(tls_client.tls_config.driver.clone());
+                match tokio::time::timeout(
+                    tls_client.tls_config.handshake_timeout,
+                    tls_connector.connect(tls_client.tls_name.clone(), stream),
+                )
+                .await
+                {
+                    Ok(Ok(stream)) => self.redis_handshake(stream).await,
+                    Ok(Err(e)) => Err(anyhow!("failed to tls handshake with {}: {e}", self.server)),
+                    Err(_) => Err(anyhow!("timeout to tls handshake with {}", self.server)),
+                }
+            } else {
+                self.redis_handshake(stream).await
+            };
         }
+
+        Err(last_err.unwrap_or_else(|| anyhow!("no ip address resolved for {}", self.server)))
     }
 
     async fn redis_handshake<S>(&self, stream: S) -> anyhow::Result<MultiplexedConnection>
@@ -192,10 +224,15 @@ impl RedisClientConfig {
         let async_config =
             AsyncConnectionConfig::new().set_response_timeout(Some(self.response_timeout));
 
-        let (conn, background) =
-            MultiplexedConnection::new_with_config(&self.db_info, stream, async_config)
-                .await
-                .map_err(|e| anyhow!("redis handshake with {} failed: {e}", self.server))?;
+        let handshake = MultiplexedConnection::new_with_config(&self.db_info, stream, async_config);
+        let (conn, background) = match tokio::time::timeout(self.response_timeout, handshake).await
+        {
+            Ok(Ok(conn)) => conn,
+            Ok(Err(e)) => {
+                return Err(anyhow!("redis handshake with {} failed: {e}", self.server));
+            }
+            Err(_) => return Err(anyhow!("timeout to redis handshake with {}", self.server)),
+        };
         tokio::spawn(background);
         Ok(conn)
     }
@@ -282,4 +319,67 @@ mod tests {
         let tls = config.tls_client.as_ref().unwrap();
         assert_eq!(tls.tls_name, ServerName::try_from("10.0.0.5").unwrap());
     }
+
+    fn silent_peer_config(port: u16) -> RedisClientConfig {
+        let mut builder = RedisClientConfigBuilder::new(UpstreamAddr::new(
+            Host::Ip(IpAddr::from_str("127.0.0.1").unwrap()),
+            port,
+        ));
+        builder.set_connect_timeout(Duration::from_millis(500));
+        builder.set_response_timeout(Duration::from_millis(200));
+        builder.build().unwrap()
+    }
+
+    #[tokio::test]
+    async fn handshake_times_out_when_peer_is_silent() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            let (sock, _) = listener.accept().await.unwrap();
+            let _hold = sock;
+            std::future::pending::<()>().await;
+        });
+
+        let config = silent_peer_config(addr.port());
+        let started = std::time::Instant::now();
+        let err = match config.connect().await {
+            Ok(_) => panic!("silent peer should not finish handshake"),
+            Err(e) => e,
+        };
+        let elapsed = started.elapsed();
+
+        assert!(
+            err.to_string().contains("timeout to redis handshake"),
+            "{err}"
+        );
+        assert!(elapsed >= Duration::from_millis(150), "{elapsed:?}");
+        assert!(elapsed < Duration::from_millis(1500), "{elapsed:?}");
+    }
+
+    #[tokio::test]
+    async fn connect_tries_next_address_after_refusal() {
+        let closed = {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let addr = listener.local_addr().unwrap();
+            drop(listener);
+            addr
+        };
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let open = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            let (sock, _) = listener.accept().await.unwrap();
+            let _hold = sock;
+            std::future::pending::<()>().await;
+        });
+
+        let config = silent_peer_config(open.port());
+        let err = match config.connect_peers(vec![closed, open]).await {
+            Ok(_) => panic!("closed then silent peer should not finish handshake"),
+            Err(e) => e,
+        };
+        assert!(
+            err.to_string().contains("timeout to redis handshake"),
+            "{err}"
+        );
+    }
 }

lib/vey-reuseport/src/lib.rs

@@ -3,12 +3,27 @@
  * SPDX-FileCopyrightText: 2026 VEY-OSS Developers.
  */
 
+use std::net::{IpAddr, SocketAddr};
+use std::path::PathBuf;
+
 use zerocopy::{Immutable, IntoBytes};
 
 pub mod quic;
 pub mod tcp;
 pub mod udp;
 
+/// Pin directory for one listener. IPv4 and IPv6 never share it: `0.0.0.0`
+/// and `::` are different sockets, and a shared map collides on worker keys.
+pub(crate) fn reuseport_pin_dir(protocol: &str, addr: SocketAddr) -> PathBuf {
+    let ip = match addr.ip() {
+        // '.' is not allowed in the pin path.
+        IpAddr::V4(ip) => format!("v4-{:08x}", u32::from(ip)),
+        IpAddr::V6(ip) => format!("v6-{}", ip.to_string().replace('.', "-")),
+    };
+    let port = addr.port();
+    PathBuf::from(format!("/sys/fs/bpf/vey-reuseport/{protocol}/{ip}_{port}"))
+}
+
 #[derive(IntoBytes, Immutable)]
 #[repr(C)]
 struct SocketId {
@@ -43,6 +58,7 @@ struct ReadOnlyData {
 mod tests {
     use super::*;
     use std::mem::size_of;
+    use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 
     #[test]
     fn bpf_struct_layout_matches_expectations() {
@@ -63,4 +79,36 @@ mod tests {
         assert_eq!(bytes.len(), size_of::<SocketId>());
         assert_eq!(bytes.len(), 8);
     }
+
+    #[test]
+    fn unspecified_v4_and_v6_do_not_share_pin_dir() {
+        let v4 = SocketAddr::from((Ipv4Addr::UNSPECIFIED, 443));
+        let v6 = SocketAddr::from((Ipv6Addr::UNSPECIFIED, 443));
+        let v4_dir = reuseport_pin_dir("tcp", v4);
+        let v6_dir = reuseport_pin_dir("tcp", v6);
+
+        assert_ne!(v4_dir, v6_dir);
+        assert!(!v4_dir.display().to_string().contains('.'));
+        assert!(!v6_dir.display().to_string().contains('.'));
+        assert!(v4_dir.ends_with("v4-00000000_443"));
+        assert!(v6_dir.ends_with("v6-::_443"));
+    }
+
+    #[test]
+    fn ipv4_localhost_does_not_share_pin_dir_with_ipv6_localhost() {
+        let v4 = SocketAddr::from((Ipv4Addr::new(0, 0, 0, 1), 443));
+        let v6 = SocketAddr::from((Ipv6Addr::LOCALHOST, 443));
+        assert_ne!(reuseport_pin_dir("udp", v4), reuseport_pin_dir("udp", v6));
+    }
+
+    #[test]
+    fn ipv4_does_not_share_pin_dir_with_mapped_ipv6() {
+        let v4 = SocketAddr::from((Ipv4Addr::LOCALHOST, 53));
+        let mapped = SocketAddr::from((Ipv4Addr::LOCALHOST.to_ipv6_mapped(), 53));
+        let v4_dir = reuseport_pin_dir("quic", v4);
+        let v6_dir = reuseport_pin_dir("quic", mapped);
+
+        assert_ne!(v4_dir, v6_dir);
+        assert!(!v6_dir.display().to_string().contains('.'));
+    }
 }

lib/vey-reuseport/src/quic.rs

@@ -3,7 +3,7 @@
  * SPDX-FileCopyrightText: 2026 VEY-OSS Developers.
  */
 
-use std::net::{IpAddr, SocketAddr};
+use std::net::SocketAddr;
 use std::os::fd::AsFd;
 use std::path::{Path, PathBuf};
 use std::{fs, ptr};
@@ -93,12 +93,7 @@ impl QuicSocketSelector {
     }
 
     pub fn new(pid: i32, generation: u16, addr: SocketAddr) -> anyhow::Result<Self> {
-        let ip = match addr.ip() {
-            IpAddr::V4(ip) => ip.to_ipv6_compatible(), // IPv4 "." is not allowed in path
-            IpAddr::V6(ip) => ip,
-        };
-        let dir = format!("/sys/fs/bpf/vey-reuseport/quic/{ip}_{}", addr.port());
-        let pin_dir = PathBuf::from(dir);
+        let pin_dir = super::reuseport_pin_dir("quic", addr);
         let quic_conn_track_path = pin_dir.join(NAME_QUIC_CONN_TRACK);
         let socket_map_path = pin_dir.join(NAME_SOCKET_MAP);