diff --git a/components/src/dynamo/sglang/publisher.py b/components/src/dynamo/sglang/publisher.py index 867e3ffa9797..9b9bead30808 100644 --- a/components/src/dynamo/sglang/publisher.py +++ b/components/src/dynamo/sglang/publisher.py @@ -163,6 +163,21 @@ def format_zmq_endpoint(endpoint_template: str, ip_address: str) -> str: return NetworkAddress(ip_address, parsed.port).to_tcp() +_WILDCARD_HOSTS = {"*", "0.0.0.0", "::"} + + +def kv_event_connect_ip(endpoint: str) -> str: + """Return the address the local subscriber uses to reach SGLang's KV event publisher. + + SGLang binds a wildcard endpoint without ZMQ_IPV6, so it listens on IPv4 only. + Loopback reaches that listener on every host, including IPv6-only hosts where the + host's own address is IPv6. + """ + if urlparse(endpoint).hostname in _WILDCARD_HOSTS: + return "127.0.0.1" + return get_local_ip_auto() + + # Note: We use SGLang's ZmqEventPublisher.offset_endpoint_port() directly # to ensure perfect alignment between publisher (SGLang) and subscriber (dynamo). # This is the same pattern used by dynamo+vLLM. @@ -384,7 +399,7 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]: raise ValueError( "sglang kv_events_config is set but missing 'endpoint'" ) - local_ip = get_local_ip_auto() + connect_ip = kv_event_connect_ip(base_ep) # Determine DP attention configuration dp_ranks = get_local_dp_rank_range(self.server_args) @@ -406,7 +421,7 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]: ) continue - zmq_ep = format_zmq_endpoint(zmq_ep, local_ip) + zmq_ep = format_zmq_endpoint(zmq_ep, connect_ip) logging.info( f"Setting up ZMQ kv event subscriber for dp_rank={dp_rank} " diff --git a/components/src/dynamo/sglang/tests/test_sglang_publisher.py b/components/src/dynamo/sglang/tests/test_sglang_publisher.py index 40af00e49a9e..ceb1e333c8fc 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_publisher.py +++ b/components/src/dynamo/sglang/tests/test_sglang_publisher.py @@ -489,6 +489,23 @@ def cleanup(self): assert metrics_task.cancelled() +@pytest.mark.parametrize( + ("endpoint", "expected"), + [ + ("tcp://*:5557", "127.0.0.1"), + ("tcp://0.0.0.0:5557", "127.0.0.1"), + ("tcp://[::]:5557", "127.0.0.1"), + ("tcp://192.0.2.10:5557", "2001:db8::7"), + ], +) +def test_kv_event_connect_ip_uses_loopback_for_wildcard_binds( + monkeypatch, endpoint, expected +): + monkeypatch.setattr(publisher_mod, "get_local_ip_auto", lambda: "2001:db8::7") + + assert publisher_mod.kv_event_connect_ip(endpoint) == expected + + def test_init_kv_event_publish_uses_worker_id_override(monkeypatch): calls = [] diff --git a/deploy/operator/internal/dynamo/backend_vllm.go b/deploy/operator/internal/dynamo/backend_vllm.go index d01a61645526..43d4f8f7ddb0 100644 --- a/deploy/operator/internal/dynamo/backend_vllm.go +++ b/deploy/operator/internal/dynamo/backend_vllm.go @@ -168,9 +168,16 @@ def _k8s_api(): ns = open(f"{SA}/namespace").read() return f"https://kubernetes.default.svc/api/v1/namespaces/{ns}/pods" +def resolve_leader_ip(): + # gethostbyname is IPv4-only; fall back to IPv6 for IPv6-only clusters. + try: + return socket.gethostbyname(host) + except socket.gaierror: + return socket.getaddrinfo(host, None, socket.AF_INET6, socket.SOCK_STREAM)[0][4][0] + def leader_pod_is_healthy(): try: - ip = socket.gethostbyname(host) + ip = resolve_leader_ip() except socket.gaierror: return False, "DNS resolution failed", None, None try: diff --git a/docs/fern/pages/kubernetes/kv-aware-routing/kv-dc-relay.md b/docs/fern/pages/kubernetes/kv-aware-routing/kv-dc-relay.md index d966bc0c17fe..ea88e6fc860e 100644 --- a/docs/fern/pages/kubernetes/kv-aware-routing/kv-dc-relay.md +++ b/docs/fern/pages/kubernetes/kv-aware-routing/kv-dc-relay.md @@ -236,8 +236,9 @@ Relay discovers direct ZMQ publishers through Kubernetes metadata and connects t advertised TCP addresses. Allow those pod-to-pod connections, worker recovery requests, and the return path to Relay's response port. Publisher ports may be dynamically allocated; opening only `5561` is insufficient. On multi-interface workers, set `DYN_EVENT_PLANE_HOST` -to a reachable pod IP if automatic selection advertises the wrong address. It changes the -advertised address, not the listener's bind address; use routable IPv4 addresses for direct ZMQ. +to a reachable IPv4 or IPv6 pod address if automatic selection advertises the wrong address. +Direct ZMQ publishers bind `0.0.0.0` for an advertised IPv4 address or `[::]` for an advertised +IPv6 address. Relay connects to the advertised address and the port assigned to that listener. Discovery RBAC, the WAN Service, optional mTLS sidecar, and the gRPC checks below are unchanged. diff --git a/docs/fern/pages/reference/components/frontend-configuration.mdx b/docs/fern/pages/reference/components/frontend-configuration.mdx index dbf985999e8b..90bc457b1fe7 100644 --- a/docs/fern/pages/reference/components/frontend-configuration.mdx +++ b/docs/fern/pages/reference/components/frontend-configuration.mdx @@ -16,7 +16,8 @@ Every CLI argument has a corresponding environment variable. The CLI argument ta ## HTTP and networking - HTTP listen address. + HTTP listen address: an IPv4 or IPv6 address, with optional brackets around IPv6. `0.0.0.0` + accepts IPv4 connections only; on IPv6-only hosts, set `::`. Environment variable: `DYN_HTTP_HOST` diff --git a/docs/fern/pages/reference/components/runtime-configuration.mdx b/docs/fern/pages/reference/components/runtime-configuration.mdx index c9f269e0c79a..04bd3bac07f4 100644 --- a/docs/fern/pages/reference/components/runtime-configuration.mdx +++ b/docs/fern/pages/reference/components/runtime-configuration.mdx @@ -210,15 +210,29 @@ Unless a field is marked environment-only, it has both a CLI flag and an environ usable non-loopback IPv4 address first, then IPv6. On Unix, automatic selection excludes down interfaces. If neither family has a usable non-loopback address, Dynamo selects IPv4 loopback, then IPv6 loopback, and falls back to `127.0.0.1` if neither is found. Remote subscribers cannot - reach a loopback address. + reach a loopback address. Automatic selection can choose an IPv4 link-local address or a bridge + address. On hosts with several interfaces, set this variable to a reachable IP address or interface + if automatic selection chooses the wrong address. Direct ZMQ publishers bind `0.0.0.0` for an advertised IPv4 address or `[::]` for an advertised IPv6 address. Discovery publishes the concrete address with the port assigned to that listener. - An explicit override must identify an address reachable by subscribers. Event-plane PUB/SUB - sockets support IPv4 and IPv6, including connections to configured ZMQ brokers. This variable - does not affect NATS or ZMQ broker mode. + An explicit override must identify an address reachable by subscribers. The IPv6 wildcard can + also accept IPv4 connections; keep ZMQ listeners on trusted networks. + + Event-plane PUB/SUB sockets support IPv4 addresses and bracketed IPv6 addresses, including + connections to configured ZMQ brokers. Hostnames retain IPv4 resolution. For an IPv6 broker, + use a bracketed address such as `tcp://[2001:db8::10]:5555`; Dynamo rejects an unbracketed IPv6 + address. This variable does not affect NATS + or ZMQ broker mode. + +On IPv6-only hosts, set `DYN_HTTP_HOST=::` on the Frontend and `DYN_SYSTEM_HOST=::` on workers. Both +listeners default to `0.0.0.0`, which accepts IPv4 connections only. The request plane and the direct +ZMQ event plane select an IPv6 address automatically when the host has no usable non-loopback IPv4 +address. + + Number of libzmq background I/O threads in the process-wide event-plane context. All event-plane PUB/SUB sockets in a frontend or worker process share this context. This setting does not change @@ -358,6 +372,15 @@ Unless a field is marked environment-only, it has both a CLI flag and an environ effective per-worker cap is the engine limit plus this queue limit. + + Listen address for the system status server on `DYN_SYSTEM_PORT`, which serves health, metrics, + and self-hosted model metadata. `0.0.0.0` accepts IPv4 connections only; on IPv6-only hosts, set + `::`. For a wildcard address, Dynamo advertises a non-loopback address from the same family; `::` + also accepts IPv4 connections, so it falls back to a non-loopback IPv4 address. If none exists, + Dynamo advertises loopback and logs a warning; remote callers such as the Frontend cannot reach + that address. + + ## Related pages diff --git a/lib/kv-router/src/services/common/zmq.rs b/lib/kv-router/src/services/common/zmq.rs index 5d5d821e8d8b..e438617e7724 100644 --- a/lib/kv-router/src/services/common/zmq.rs +++ b/lib/kv-router/src/services/common/zmq.rs @@ -59,6 +59,7 @@ impl ZmqSocket { } pub(crate) fn connect(&self, endpoint: &str) -> Result<()> { + set_ipv6_for(self.socket(), endpoint)?; self.socket().connect(endpoint)?; Ok(()) } @@ -161,6 +162,13 @@ impl ZmqSocket { } } +/// libzmq resolves TCP hosts as IPv4 unless `ZMQ_IPV6` is set, and snapshots the option per +/// bind or connect. [`validate_endpoint`] rejects unbracketed IPv6 literals. +fn set_ipv6_for(socket: &zmq::Socket, endpoint: &str) -> Result<()> { + socket.set_ipv6(endpoint.starts_with("tcp://["))?; + Ok(()) +} + fn configure_common_socket(socket: &zmq::Socket) -> Result<()> { socket.set_linger(ZMQ_LINGER_MS)?; socket.set_reconnect_ivl(ZMQ_RECONNECT_IVL_MS)?; @@ -231,6 +239,7 @@ pub(crate) fn connect_sub_socket(endpoint: &str) -> Result { pub(crate) fn connect_dealer_socket(endpoint: &str) -> Result { Ok(Arc::new(Mutex::new(build_socket(zmq::DEALER, |socket| { configure_bidirectional_socket(socket)?; + set_ipv6_for(socket, endpoint)?; socket.connect(endpoint)?; Ok(()) })?))) @@ -244,6 +253,7 @@ pub(crate) fn connect_dealer_socket(endpoint: &str) -> Result { pub(crate) fn create_bound_pub_socket(endpoint: &str) -> Result { build_socket(zmq::PUB, |socket| { configure_send_socket(socket)?; + set_ipv6_for(socket, endpoint)?; socket.bind(endpoint)?; Ok(()) }) @@ -304,3 +314,40 @@ pub(crate) fn validate_endpoint(endpoint: &str) -> Result<()> { )), } } + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + #[tokio::test] + async fn bracketed_ipv6_endpoints_connect() { + if let Err(error) = std::net::TcpListener::bind("[::1]:0") { + eprintln!("Skipping IPv6 ZMQ test: {error}"); + return; + } + let mut publisher = create_bound_pub_socket("tcp://[::1]:*").unwrap(); + let endpoint = publisher.socket().get_last_endpoint().unwrap().unwrap(); + validate_endpoint(&endpoint).unwrap(); + let mut subscriber = create_sub_socket(b"").unwrap(); + subscriber.connect(&endpoint).unwrap(); + + tokio::time::timeout(Duration::from_secs(5), async { + loop { + publisher + .send_multipart(vec![b"ipv6".to_vec()]) + .await + .unwrap(); + if let Ok(frames) = + tokio::time::timeout(Duration::from_millis(50), subscriber.recv_multipart()) + .await + { + assert_eq!(frames.unwrap(), vec![b"ipv6".to_vec()]); + break; + } + } + }) + .await + .expect("IPv6 SUB socket should receive the message"); + } +} diff --git a/lib/llm/src/http/service/service_v2.rs b/lib/llm/src/http/service/service_v2.rs index 28f91b230037..c79ceb28594c 100644 --- a/lib/llm/src/http/service/service_v2.rs +++ b/lib/llm/src/http/service/service_v2.rs @@ -45,7 +45,7 @@ use dynamo_runtime::metrics::{ tokio_perf::{ensure_tokio_perf_metrics_registered_prometheus, tokio_metrics_and_canary_loop}, transport_metrics::ensure_transport_metrics_registered_prometheus, }; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use tokio::sync::Notify; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -926,9 +926,7 @@ impl HttpService { Use run()/spawn() (which bind internally) when enable_tls is set." )); } - let addr: SocketAddr = address - .parse() - .map_err(|e| anyhow::anyhow!("Invalid address '{}': {}", address, e))?; + let addr = listen_socket_addr(&self.host, self.port)?; let cert_path = self .tls_cert_path .as_ref() @@ -1007,9 +1005,7 @@ impl HttpService { let listener = match listener { Some(l) => l, None => { - let addr: SocketAddr = address - .parse() - .map_err(|e| anyhow::anyhow!("Invalid address '{}': {}", address, e))?; + let addr = listen_socket_addr(&self.host, self.port)?; bind_listener(addr).map_err(|e| { tracing::error!( protocol = %protocol, @@ -1075,11 +1071,11 @@ impl HttpService { let Some(rl_router) = self.rl_router.clone() else { return Ok(()); }; - let rl_addr = format!("{}:{}", self.host, self.rl_port); + let rl_addr = listen_socket_addr(&self.host, self.rl_port)?; // Bind eagerly and fail fast: when RL discovery is enabled, a bind failure // should abort service startup rather than silently leave RL discovery // unavailable while the main HTTP service keeps running. - let listener = tokio::net::TcpListener::bind(&rl_addr).await.map_err(|e| { + let listener = tokio::net::TcpListener::bind(rl_addr).await.map_err(|e| { tracing::error!( address = %rl_addr, error = %e, @@ -1155,6 +1151,20 @@ fn listen_backlog() -> u32 { parse_listen_backlog(std::env::var(env_llm::DYN_HTTP_LISTEN_BACKLOG)) } +/// Build the listen address from an IP literal host. IPv6 brackets are optional, +/// so `::` and `[::]` are equivalent. +fn listen_socket_addr(host: &str, port: u16) -> Result { + let literal = host.trim(); + let literal = literal + .strip_prefix('[') + .and_then(|inner| inner.strip_suffix(']')) + .unwrap_or(literal); + let ip: IpAddr = literal + .parse() + .map_err(|e| anyhow::anyhow!("Invalid address '{host}:{port}': {e}"))?; + Ok(SocketAddr::new(ip, port)) +} + /// `tokio::net::TcpListener::bind` listens with a backlog of 128. A few thousand /// clients connecting within seconds overflow that, and overflowed connections /// are delayed or, depending on host TCP settings, fail. @@ -2774,6 +2784,30 @@ mod tests { assert_eq!(peer, client.local_addr().unwrap()); } + #[test] + fn test_listen_socket_addr_accepts_ip_literals() { + for (host, expected) in [ + ("0.0.0.0", "0.0.0.0:8000"), + ("127.0.0.1", "127.0.0.1:8000"), + ("::", "[::]:8000"), + ("[::]", "[::]:8000"), + (" [::1] ", "[::1]:8000"), + ] { + assert_eq!( + listen_socket_addr(host, 8000).unwrap(), + expected.parse::().unwrap(), + "{host}" + ); + } + for host in ["localhost", "[::", "::1]", ""] { + let error = listen_socket_addr(host, 8000).unwrap_err().to_string(); + assert!( + error.starts_with(&format!("Invalid address '{host}:8000'")), + "{host}: {error}" + ); + } + } + #[test] fn test_sse_keep_alive_for_deferred_response() { let configured = Duration::from_millis(5000); diff --git a/lib/llm/src/utils/zmq.rs b/lib/llm/src/utils/zmq.rs index 337d37e6c6e4..3d107f838c33 100644 --- a/lib/llm/src/utils/zmq.rs +++ b/lib/llm/src/utils/zmq.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use anyhow::Result; +use dynamo_runtime::transports::zmq::ipv6_option_for; use futures::SinkExt; use tmq::{ Context, Multipart, SocketBuilder, @@ -56,6 +57,7 @@ where pub(crate) async fn connect_sub_socket(endpoint: &str, topic: Option<&str>) -> Result { let ctx = Context::new(); let socket = configure_receive_builder(subscribe(&ctx)) + .set_ipv6(ipv6_option_for(endpoint)?) .connect(endpoint)? .subscribe(topic.unwrap_or("").as_bytes())?; Ok(socket) @@ -64,20 +66,26 @@ pub(crate) async fn connect_sub_socket(endpoint: &str, topic: Option<&str>) -> R #[cfg_attr(not(feature = "block-manager"), allow(dead_code))] pub(crate) async fn bind_pub_socket(endpoint: &str) -> Result { let ctx = Context::new(); - let socket = configure_send_builder(publish(&ctx)).bind(endpoint)?; + let socket = configure_send_builder(publish(&ctx)) + .set_ipv6(ipv6_option_for(endpoint)?) + .bind(endpoint)?; Ok(Arc::new(Mutex::new(socket))) } pub(crate) async fn bind_pull_socket(endpoint: &str) -> Result { let ctx = Context::new(); - let socket = configure_receive_builder(pull(&ctx)).bind(endpoint)?; + let socket = configure_receive_builder(pull(&ctx)) + .set_ipv6(ipv6_option_for(endpoint)?) + .bind(endpoint)?; Ok(socket) } #[cfg(test)] pub(crate) async fn connect_push_socket(endpoint: &str) -> Result { let ctx = Context::new(); - let socket = configure_send_builder(tmq::push::push(&ctx)).connect(endpoint)?; + let socket = configure_send_builder(tmq::push::push(&ctx)) + .set_ipv6(ipv6_option_for(endpoint)?) + .connect(endpoint)?; Ok(socket) } @@ -105,3 +113,45 @@ where socket.send(Multipart::from(frames)).await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use futures::StreamExt; + use tmq::AsZmqSocket; + + #[tokio::test] + async fn bracketed_ipv6_endpoints_connect() { + if let Err(error) = std::net::TcpListener::bind("[::1]:0") { + eprintln!("Skipping IPv6 ZMQ test: {error}"); + return; + } + let mut pull = bind_pull_socket("tcp://[::1]:*").await.unwrap(); + let endpoint = pull.get_socket().get_last_endpoint().unwrap().unwrap(); + let mut push = connect_push_socket(&endpoint).await.unwrap(); + + send_multipart_direct(&mut push, vec![b"ipv6".to_vec()]) + .await + .unwrap(); + let received = tokio::time::timeout(std::time::Duration::from_secs(5), pull.next()) + .await + .expect("IPv6 PULL socket should receive the message") + .unwrap() + .unwrap(); + assert_eq!(multipart_message(received), vec![b"ipv6".to_vec()]); + } + + #[tokio::test] + async fn unbracketed_ipv6_endpoints_are_rejected() { + let error = connect_sub_socket("tcp://::1:5555", None) + .await + .err() + .expect("unbracketed IPv6 endpoint should be rejected"); + assert!( + error + .to_string() + .contains("IPv6 addresses must be bracketed"), + "{error}" + ); + } +} diff --git a/lib/runtime/src/system_status_server.rs b/lib/runtime/src/system_status_server.rs index 97d744a9a810..41c7d8c7be9a 100644 --- a/lib/runtime/src/system_status_server.rs +++ b/lib/runtime/src/system_status_server.rs @@ -67,7 +67,19 @@ impl SystemStatusServerInfo { /// used so IPv4 and IPv6 are never combined. pub fn advertised_socket_addr(&self) -> SocketAddr { match advertised_socket_addr(self.socket_addr, &DefaultIpResolver) { - Ok(address) => address, + Ok(address) => { + if self.socket_addr.ip().is_unspecified() && address.ip().is_loopback() { + static WARNED: std::sync::Once = std::sync::Once::new(); + WARNED.call_once(|| { + tracing::warn!( + bind = %self.socket_addr, + advertised = %address, + "System status server found no non-loopback address in the bound family, so remote callers cannot reach the advertised address; on IPv6-only hosts, set DYN_SYSTEM_HOST=::" + ); + }); + } + address + } Err(error) => { let fallback = match self.socket_addr.ip() { IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST), diff --git a/lib/runtime/src/transports/event_plane/mod.rs b/lib/runtime/src/transports/event_plane/mod.rs index ea3a6f235486..649c63c97d7c 100644 --- a/lib/runtime/src/transports/event_plane/mod.rs +++ b/lib/runtime/src/transports/event_plane/mod.rs @@ -60,7 +60,14 @@ fn event_plane_host_from_env() -> Result { fn event_plane_host_from_env_with_resolver(resolver: &R) -> Result { let Some(host) = host_override_from_env(DYN_EVENT_PLANE_HOST)? else { - return Ok(resolve_local_host(resolver)?.advertise_ip()); + let resolved = resolve_local_host(resolver)?; + if resolved.used_loopback_fallback() { + tracing::warn!( + address = %resolved.advertise_ip(), + "No usable non-loopback IP address found; set DYN_EVENT_PLANE_HOST to an address reachable by subscribers" + ); + } + return Ok(resolved.advertise_ip()); }; resolve_host_or_interface(&host, resolver) @@ -494,6 +501,11 @@ impl EventPublisher { .context("Failed to create Tokio runtime for ZMQ")?; rt.block_on(ZmqPubTransport::bind(bind_endpoint, &topic)) + .with_context(|| { + format!( + "Failed to bind direct ZMQ publisher at {bind_endpoint} for advertised address {advertised_host}; check DYN_EVENT_PLANE_HOST and host IP support" + ) + }) } }) .join() @@ -963,7 +975,7 @@ mod tests { use crate::utils::ip_resolver::test_support::StubResolver; #[test] - fn direct_zmq_automatic_host_prefers_non_loopback_then_ipv4() { + fn direct_zmq_automatic_host_selection() { let mut resolver = StubResolver::not_found(); resolver.interfaces = vec![ ("lo", "127.0.0.1".parse().unwrap()), @@ -990,40 +1002,13 @@ mod tests { ); } - struct EventPlaneHostResolver { - ipv4: Option, - ipv6: Option, - interfaces: Vec<(String, std::net::IpAddr)>, - } - - impl IpResolver for EventPlaneHostResolver { - fn local_ip(&self) -> std::result::Result { - self.ipv4 - .ok_or(local_ip_address::Error::LocalIpAddressNotFound) - } - - fn local_ipv6(&self) -> std::result::Result { - self.ipv6 - .ok_or(local_ip_address::Error::LocalIpAddressNotFound) - } - - fn list_afinet_netifas( - &self, - ) -> std::result::Result, local_ip_address::Error> { - Ok(self.interfaces.clone()) - } - } - #[test] fn direct_zmq_advertise_host_from_env_resolves_ips_and_interfaces() { - let resolver = EventPlaneHostResolver { - ipv4: Some("192.0.2.1".parse().unwrap()), - ipv6: None, - interfaces: vec![ - ("ib0".to_string(), "192.0.2.20".parse().unwrap()), - ("ib6".to_string(), "2001:db8::20".parse().unwrap()), - ], - }; + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![ + ("ib0", "192.0.2.20".parse().unwrap()), + ("ib6", "2001:db8::20".parse().unwrap()), + ]; assert_eq!( temp_env::with_vars([(DYN_EVENT_PLANE_HOST, Some(" 192.0.2.10 "))], || { @@ -1059,11 +1044,8 @@ mod tests { #[test] fn direct_zmq_advertise_host_falls_back_to_ipv6_and_rejects_wildcards() { - let resolver = EventPlaneHostResolver { - ipv4: None, - ipv6: Some("2001:db8::1".parse().unwrap()), - interfaces: vec![("eth0".to_string(), "2001:db8::1".parse().unwrap())], - }; + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![("eth0", "2001:db8::1".parse().unwrap())]; assert_eq!( temp_env::with_vars([(DYN_EVENT_PLANE_HOST, None::<&str>)], || { event_plane_host_from_env_with_resolver(&resolver) @@ -1117,6 +1099,12 @@ mod tests { #[case("[::1]")] #[tokio::test] async fn direct_zmq_publisher_serves_advertised_endpoint(#[case] host: &str) { + if host == "[::1]" + && let Err(error) = std::net::TcpListener::bind("[::1]:0") + { + eprintln!("Skipping IPv6 publisher test: {error}"); + return; + } temp_env::async_with_vars( [ (DYN_EVENT_PLANE_HOST, Some(host)), diff --git a/lib/runtime/src/transports/event_plane/zmq_transport.rs b/lib/runtime/src/transports/event_plane/zmq_transport.rs index f5db00bc08a5..48d20f331b9e 100644 --- a/lib/runtime/src/transports/event_plane/zmq_transport.rs +++ b/lib/runtime/src/transports/event_plane/zmq_transport.rs @@ -80,6 +80,7 @@ use super::codec::{Codec, MsgpackCodec}; use super::frame::Frame; use super::transport::{EventTransportRx, EventTransportTx, WireStream}; use crate::discovery::EventTransportKind; +use crate::transports::zmq::ipv6_option_for; fn socket_limit_guidance(raw_errno: Option, guidance: &'static str) -> Option<&'static str> { (raw_errno == Some(libc::EMFILE)).then_some(guidance) @@ -114,14 +115,33 @@ fn bind_tmq_socket(builder: SocketBuilder, endpoint: &str) -> Result where T: tmq::FromZmqSocket, { - builder.bind(endpoint).map_err(map_socket_creation_error) + builder + .set_ipv6(ipv6_option_for(endpoint)?) + .bind(endpoint) + .map_err(map_socket_creation_error) } fn connect_tmq_socket(builder: SocketBuilder, endpoint: &str) -> Result where T: tmq::FromZmqSocket, { - builder.connect(endpoint).map_err(map_socket_creation_error) + builder + .set_ipv6(ipv6_option_for(endpoint)?) + .connect(endpoint) + .map_err(map_socket_creation_error) +} + +fn connect_zmq_socket(socket: &impl AsZmqSocket, endpoint: &str) -> Result<()> { + let ipv6 = ipv6_option_for(endpoint)?; + let socket = socket.get_socket(); + // ZMQ snapshots this option per connection; keep hostname resolution on IPv4. + socket + .set_ipv6(ipv6) + .with_context(|| format!("Failed to set ZMQ_IPV6 for {endpoint}"))?; + socket + .connect(endpoint) + .with_context(|| format!("Failed to connect ZMQ socket to {endpoint}"))?; + Ok(()) } fn configure_publish_builder(builder: SocketBuilder) -> SocketBuilder @@ -129,7 +149,6 @@ where T: tmq::FromZmqSocket, { builder - .set_ipv6(true) .set_sndhwm(ZMQ_SNDHWM) .set_sndtimeo(ZMQ_SNDTIMEOUT_MS) } @@ -148,10 +167,7 @@ fn configure_subscribe_builder_with_hwm( where T: tmq::FromZmqSocket, { - builder - .set_ipv6(true) - .set_rcvhwm(rcvhwm) - .set_rcvtimeo(ZMQ_RCVTIMEOUT_MS) + builder.set_rcvhwm(rcvhwm).set_rcvtimeo(ZMQ_RCVTIMEOUT_MS) } /// Keeps a received ZMQ message alive for as long as any derived `Bytes` exists. @@ -243,7 +259,7 @@ impl ZmqPubTransport { let socket = connect_tmq_socket(configure_publish_builder(publish(&ctx)), first_endpoint)?; for endpoint in endpoints { - socket.get_socket().connect(endpoint)?; + connect_zmq_socket(&socket, endpoint)?; tracing::debug!(endpoint = %endpoint, "ZMQ PUB connected to broker XSUB"); } @@ -328,7 +344,7 @@ impl DynamicZmqSubSocket { /// Connect this SUB socket to one more publisher endpoint. pub fn add_endpoint(&mut self, endpoint: &str) -> Result<()> { - self.socket.get_socket().connect(endpoint)?; + connect_zmq_socket(&self.socket, endpoint)?; Ok(()) } @@ -532,7 +548,7 @@ impl ZmqSubTransport { let endpoint_count = endpoints.len(); let socket = Self::connect_socket(first_endpoint, topic)?; for endpoint in endpoint_iter { - socket.get_socket().connect(endpoint)?; + connect_zmq_socket(&socket, endpoint)?; } tracing::info!( @@ -593,7 +609,7 @@ impl ZmqSubTransport { .subscribe(topic.as_bytes())?; for endpoint in endpoints_iter { - socket.get_socket().connect(endpoint)?; + connect_zmq_socket(&socket, endpoint)?; tracing::debug!(endpoint = %endpoint, "ZMQ SUB connected to endpoint"); } @@ -783,6 +799,32 @@ mod tests { assert_eq!(map_socket_creation_error(error).to_string(), original); } + #[tokio::test] + async fn unbracketed_ipv6_endpoints_fail_before_connecting() { + let error = ZmqSubTransport::connect_broker("tcp://::1:5555", "topic") + .await + .err() + .expect("unbracketed IPv6 endpoint should be rejected"); + assert!( + error + .to_string() + .contains("IPv6 addresses must be bracketed"), + "{error}" + ); + } + + #[tokio::test] + async fn failed_additional_connection_names_endpoint() { + let endpoint = format!("inproc://dynamo-zmq-connect-context-{}", std::process::id()); + let mut subscriber = + DynamicZmqSubSocket::connect_with_rcvhwm(&endpoint, "topic", ZMQ_RCVHWM).unwrap(); + let error = subscriber.add_endpoint("tcp://missing-port").unwrap_err(); + assert!( + format!("{error:#}").contains("tcp://missing-port"), + "{error:#}" + ); + } + async fn send_raw(publisher: &ZmqPubTransport, frames: Vec>) { publisher .socket @@ -869,6 +911,7 @@ mod tests { let (publisher, endpoint) = ZmqPubTransport::bind("tcp://127.0.0.1:0", topic) .await .expect("Failed to create publisher"); + assert!(endpoint.starts_with("tcp://127.0.0.1:")); tokio::time::sleep(Duration::from_millis(100)).await; @@ -1062,24 +1105,98 @@ mod tests { ); } + #[tokio::test] + async fn broker_hostname_connections_reach_ipv4_peers() { + let topic = "broker-hostname"; + let ctx = shared_zmq_context().unwrap(); + let mut broker_sub = subscribe(&ctx) + .set_ipv6(false) + .bind("tcp://127.0.0.1:*") + .unwrap() + .subscribe(topic.as_bytes()) + .unwrap(); + let xsub_endpoint = broker_sub + .get_socket() + .get_last_endpoint() + .unwrap() + .unwrap() + .replace("127.0.0.1", "localhost"); + let publisher = ZmqPubTransport::connect(&xsub_endpoint, topic) + .await + .unwrap(); + // Some hosts have no IPv6 localhost entry; still verify the DNS family. + assert!( + !publisher + .socket + .lock() + .await + .get_socket() + .is_ipv6() + .unwrap() + ); + let (broker_pub, xpub_endpoint) = ZmqPubTransport::bind("tcp://127.0.0.1:0", topic) + .await + .unwrap(); + let subscriber = ZmqSubTransport::connect_broker( + &xpub_endpoint.replace("127.0.0.1", "localhost"), + topic, + ) + .await + .unwrap(); + let mut stream = subscriber.subscribe(topic).await.unwrap(); + let encoded = encoded_event(topic, 101, 1); + + timeout(Duration::from_secs(5), async { + loop { + publisher.publish(topic, encoded.clone()).await.unwrap(); + if let Ok(frames) = timeout(Duration::from_millis(25), broker_sub.next()).await { + broker_pub + .socket + .lock() + .await + .send(frames.unwrap().unwrap()) + .await + .unwrap(); + } + if let Ok(event) = timeout(Duration::from_millis(25), stream.next()).await { + assert_eq!(event.unwrap().unwrap(), encoded); + break; + } + } + }) + .await + .expect("hostname connections should deliver through IPv4-only peers"); + } + #[rstest::rstest] - #[case("inproc")] - #[case("tcp://127.0.0.1:0")] - #[case("tcp://[::1]:0")] + #[case("inproc", "inproc")] + #[case("tcp://127.0.0.1:0", "tcp://[::1]:0")] + #[case("tcp://[::1]:0", "tcp://127.0.0.1:0")] #[tokio::test] - async fn dynamic_sub_socket_adds_and_removes_publishers(#[case] endpoint: &str) { + async fn dynamic_sub_socket_adds_and_removes_publishers( + #[case] endpoint_a: &str, + #[case] endpoint_b: &str, + ) { + if (ipv6_option_for(endpoint_a).unwrap() || ipv6_option_for(endpoint_b).unwrap()) + && let Err(error) = std::net::TcpListener::bind("[::1]:0") + { + eprintln!("Skipping mixed IPv4/IPv6 subscriber test: {error}"); + return; + } let process = std::process::id(); - let (endpoint_a, endpoint_b) = if endpoint == "inproc" { + let (endpoint_a, endpoint_b) = if endpoint_a == "inproc" { ( format!("inproc://dynamo-zmq-dynamic-a-{process}"), format!("inproc://dynamo-zmq-dynamic-b-{process}"), ) } else { - (endpoint.to_string(), endpoint.to_string()) + (endpoint_a.to_string(), endpoint_b.to_string()) }; let topic = "dynamic-subscriber"; let (publisher_a, endpoint_a) = ZmqPubTransport::bind(&endpoint_a, topic).await.unwrap(); let (publisher_b, endpoint_b) = ZmqPubTransport::bind(&endpoint_b, topic).await.unwrap(); + let endpoint_a = endpoint_a.replace("127.0.0.1", "localhost"); + let endpoint_b = endpoint_b.replace("127.0.0.1", "localhost"); let mut subscriber = DynamicZmqSubSocket::connect_with_rcvhwm(&endpoint_a, topic, ZMQ_RCVHWM).unwrap(); subscriber.add_endpoint(&endpoint_b).unwrap(); diff --git a/lib/runtime/src/transports/zmq.rs b/lib/runtime/src/transports/zmq.rs index e6a07c5b27a7..fae7243544e7 100644 --- a/lib/runtime/src/transports/zmq.rs +++ b/lib/runtime/src/transports/zmq.rs @@ -31,6 +31,28 @@ use tokio_util::sync::CancellationToken; pub type MultipartMessage = Vec>; +/// Returns whether `endpoint` needs `ZMQ_IPV6`. +/// +/// Set the option before each bind or connect; libzmq applies it per connection. +/// Rejects unbracketed IPv6 literals: without `ZMQ_IPV6`, libzmq resolves them +/// as IPv4 and retries the connection forever without reporting an error. +pub fn ipv6_option_for(endpoint: &str) -> Result { + let Some(address) = endpoint.strip_prefix("tcp://") else { + return Ok(false); + }; + if address.starts_with('[') { + return Ok(true); + } + if let Some((host, _port)) = address.rsplit_once(':') + && host.parse::().is_ok() + { + anyhow::bail!( + "Invalid ZMQ endpoint '{endpoint}': IPv6 addresses must be bracketed, for example tcp://[{host}]:" + ); + } + Ok(false) +} + // Core message types #[derive(Debug, Clone, Serialize, Deserialize)] enum ControlMessage { @@ -354,6 +376,30 @@ mod tests { use super::*; use tokio::time::timeout; + #[test] + fn ipv6_option_follows_endpoint_address_family() { + for (endpoint, expected) in [ + ("tcp://[::1]:5555", true), + ("tcp://[::]:*", true), + ("tcp://[2001:db8::10]:5555", true), + ("tcp://127.0.0.1:5555", false), + ("tcp://0.0.0.0:*", false), + ("tcp://localhost:5555", false), + ("tcp://*:5555", false), + ("inproc://events", false), + ] { + assert_eq!(ipv6_option_for(endpoint).unwrap(), expected, "{endpoint}"); + } + + for endpoint in ["tcp://::1:5555", "tcp://2001:db8::10:5555"] { + let error = ipv6_option_for(endpoint).unwrap_err().to_string(); + assert!( + error.contains("IPv6 addresses must be bracketed"), + "{endpoint}: {error}" + ); + } + } + #[tokio::test] async fn test_basic_communication() -> Result<()> { let context = TmqContext::new(); diff --git a/lib/runtime/src/utils/ip_resolver.rs b/lib/runtime/src/utils/ip_resolver.rs index 024ee2ba42fd..ba90b3594a0e 100644 --- a/lib/runtime/src/utils/ip_resolver.rs +++ b/lib/runtime/src/utils/ip_resolver.rs @@ -319,7 +319,9 @@ fn resolve_wildcard( } /// Resolve an advertisement address that is served by an existing listener. -/// Wildcard listeners stay in their bound family. +/// Wildcard listeners prefer their bound family. An IPv6 wildcard listener also +/// accepts IPv4 by default, so it falls back to a non-loopback IPv4 address on +/// hosts without one for IPv6. An IPv4 wildcard never advertises IPv6. pub(crate) fn resolve_advertise_ip_for_bind( bind_ip: IpAddr, resolver: &R, @@ -331,8 +333,13 @@ pub(crate) fn resolve_advertise_ip_for_bind( } let candidates = local_candidates(resolver)?; + let dual_stack_fallback = bind_ip + .is_ipv6() + .then(|| candidates.other_non_loopback(bind_ip)) + .flatten(); Ok(candidates .non_loopback_for(bind_ip) + .or(dual_stack_fallback) .or_else(|| candidates.loopback_for(bind_ip)) .unwrap_or_else(|| loopback_for(bind_ip))) } @@ -839,6 +846,32 @@ mod tests { ); } + #[test] + fn ipv6_wildcard_advertisement_falls_back_to_ipv4() { + let mut resolver = StubResolver::not_found(); + resolver.interfaces = vec![ + ("lo", ip("127.0.0.1")), + ("lo", ip("::1")), + ("eth0", ip("192.0.2.20")), + ]; + assert_eq!( + resolve_advertise_ip_for_bind(ip("::"), &resolver).unwrap(), + ip("192.0.2.20") + ); + + resolver.interfaces.push(("eth0", ip("2001:db8::20"))); + assert_eq!( + resolve_advertise_ip_for_bind(ip("::"), &resolver).unwrap(), + ip("2001:db8::20") + ); + + resolver.interfaces.retain(|(name, _)| *name == "lo"); + assert_eq!( + resolve_advertise_ip_for_bind(ip("::"), &resolver).unwrap(), + ip("::1") + ); + } + #[test] fn mapped_interface_addresses_use_ipv4_classification() { let mut resolver = StubResolver::not_found();