From 4057efdabc099da32d2e30090675375f845512d3 Mon Sep 17 00:00:00 2001 From: torinnd <81178324+torinnd@users.noreply.github.com> Date: Wed, 2 Sep 2026 10:45:55 +0100 Subject: [PATCH 1/2] listeners: preserve distinct sockets across graceful upgrades Add ListenerConfig::fd_transfer_id so listeners sharing a configured address, including repeated port-zero and SO_REUSEPORT listeners, use distinct fd-table keys and retain their sockets across same-version graceful upgrades. Fixed TCP, Unix, and unidentified listeners retain their existing keys. Validate IDs during listener construction and return BindError for empty, whitespace-containing, or duplicate IDs. Track keys registered by the current process so an unidentified duplicate address is a bind error rather than a second owner of the same raw fd; no upstream test or example relies on that sharing. Mixed-version upgrades may rebind identified listeners because old binaries do not understand the new keys. Tests cover build-time validation, distinct routing, duplicate IDs, legacy keys, and a real SCM_RIGHTS round trip. --- pingora-core/src/listeners/l4.rs | 70 ++++++-- pingora-core/src/listeners/mod.rs | 190 ++++++++++++++++++++- pingora-core/src/server/transfer_fd/mod.rs | 21 +++ pingora-core/src/services/listening.rs | 2 +- 4 files changed, 263 insertions(+), 20 deletions(-) diff --git a/pingora-core/src/listeners/l4.rs b/pingora-core/src/listeners/l4.rs index 5532b635e..bf8bb9287 100644 --- a/pingora-core/src/listeners/l4.rs +++ b/pingora-core/src/listeners/l4.rs @@ -16,6 +16,7 @@ use log::debug; use log::warn; use pingora_error::{ + Error, ErrorType::{AcceptError, BindError}, OrErr, Result, }; @@ -47,14 +48,14 @@ use crate::server::ListenFds; #[cfg(unix)] use std::sync::LazyLock; -/// Per-address async lock map for serializing the check-bind-insert sequence +/// Per-key async lock map for serializing the check-bind-insert sequence /// in [`ListenerEndpointBuilder::listen`]. /// /// With `ListenFds` using a synchronous `parking_lot::Mutex`, the lock cannot /// be held across `bind().await`. This global map ensures that only one task at /// a time can be in the process of looking up, binding, and inserting a given -/// address — preventing two concurrent callers from both seeing "not found" and -/// racing to bind the same address. +/// key — preventing two concurrent callers from both seeing "not found" and +/// racing to bind the same key. #[cfg(unix)] static BIND_LOCKS: LazyLock>>> = LazyLock::new(flurry::HashMap::new); @@ -89,6 +90,11 @@ impl ServerAddress { _ => None, } } + + pub(crate) fn fd_transfer_key(&self, id: Option<&str>) -> String { + let addr = self.as_ref(); + id.map_or_else(|| addr.to_string(), |id| format!("{addr}#id={id}")) + } } /// TCP socket configuration options, this is used for setting options on @@ -307,6 +313,8 @@ pub struct ListenerEndpoint { #[derive(Default)] pub struct ListenerEndpointBuilder { listen_addr: Option, + #[cfg(unix)] + fd_transfer_key: Option, #[cfg(feature = "connection_filter")] connection_filter: Option>, } @@ -315,6 +323,8 @@ impl ListenerEndpointBuilder { pub fn new() -> ListenerEndpointBuilder { Self { listen_addr: None, + #[cfg(unix)] + fd_transfer_key: None, #[cfg(feature = "connection_filter")] connection_filter: None, } @@ -325,6 +335,12 @@ impl ListenerEndpointBuilder { self } + #[cfg(unix)] + pub(crate) fn fd_transfer_key(&mut self, key: String) -> &mut Self { + self.fd_transfer_key = Some(key); + self + } + #[cfg(feature = "connection_filter")] pub fn connection_filter(&mut self, filter: Arc) -> &mut Self { self.connection_filter = Some(filter); @@ -338,36 +354,56 @@ impl ListenerEndpointBuilder { .expect("Tried to listen with no addr specified"); let listener = if let Some(fds_table) = fds { - let addr_str = listen_addr.as_ref(); - - // Acquire a per-address async lock so that only one task at a + let key = self + .fd_transfer_key + .as_deref() + .unwrap_or_else(|| listen_addr.as_ref()); + // Acquire a per-key async lock so that only one task at a // time can go through the check-bind-insert sequence for a given - // address. The flurry guard is dropped before the await so its + // key. The flurry guard is dropped before the await so its // !Send pointer does not cross an await point. - let addr_lock = { + let key_lock = { let guard = BIND_LOCKS.pin(); - match guard.get(addr_str) { + match guard.get(key) { Some(existing) => existing.clone(), None => { - let new_lock = Arc::new(tokio::sync::Mutex::new(())); - match guard.try_insert(addr_str.to_string(), new_lock.clone()) { + let lock = Arc::new(tokio::sync::Mutex::new(())); + match guard.try_insert(key.to_string(), lock.clone()) { Ok(inserted) => inserted.clone(), Err(e) => e.current.clone(), } } } }; - let _guard = addr_lock.lock().await; - - let existing_fd = fds_table.lock().get(addr_str).copied(); + let _guard = key_lock.lock().await; + + let existing_fd = { + let fds = fds_table.lock(); + if fds.is_local(key) { + return Error::e_explain( + BindError, + format!("duplicate listener transfer identity {key}"), + ); + } + fds.get(key).copied() + }; if let Some(fd) = existing_fd { - from_raw_fd(&listen_addr, fd)? + let listener = from_raw_fd(&listen_addr, fd)?; + fds_table.lock().mark_local(key); + listener } else { let listener = bind(&listen_addr).await?; - fds_table + if fds_table .lock() - .add(addr_str.to_string(), listener.as_raw_fd()); + .try_add(key.to_string(), listener.as_raw_fd()) + .is_err() + { + return Error::e_explain( + BindError, + format!("duplicate listener transfer identity {key}"), + ); + } listener } } else { diff --git a/pingora-core/src/listeners/mod.rs b/pingora-core/src/listeners/mod.rs index 0d350493a..c3e9531c7 100644 --- a/pingora-core/src/listeners/mod.rs +++ b/pingora-core/src/listeners/mod.rs @@ -80,7 +80,7 @@ use crate::protocols::{l4::socket::SocketAddr, tls::TlsRef, Stream}; use crate::server::ListenFds; use async_trait::async_trait; -use pingora_error::Result; +use pingora_error::{Error, ErrorType::BindError, Result}; use std::{any::Any, fs::Permissions, sync::Arc}; use l4::{ListenerEndpoint, Stream as L4Stream}; @@ -165,6 +165,7 @@ struct TransportStackBuilder { l4: ServerAddress, tls: Option, l4_buffer: L4BufferSettings, + fd_transfer_id: Option, #[cfg(feature = "connection_filter")] connection_filter: Option>, pre_tls_callback: Option, @@ -175,9 +176,21 @@ impl TransportStackBuilder { &mut self, #[cfg(unix)] upgrade_listeners: Option, ) -> Result { + if self + .fd_transfer_id + .as_ref() + .is_some_and(|id| id.is_empty() || id.bytes().any(|byte| byte.is_ascii_whitespace())) + { + return Error::e_explain( + BindError, + "fd transfer ID must be non-empty and contain no ASCII whitespace", + ); + } let mut builder = ListenerEndpoint::builder(); builder.listen_addr(self.l4.clone()); + #[cfg(unix)] + builder.fd_transfer_key(self.l4.fd_transfer_key(self.fd_transfer_id.as_deref())); #[cfg(feature = "connection_filter")] if let Some(filter) = &self.connection_filter { @@ -208,6 +221,7 @@ pub struct ListenerConfig { l4: ServerAddress, tls: Option, l4_buffer: L4BufferSettings, + fd_transfer_id: Option, } impl ListenerConfig { @@ -217,6 +231,7 @@ impl ListenerConfig { l4: ServerAddress::Tcp(addr.into(), None), tls: None, l4_buffer: L4BufferSettings::default(), + fd_transfer_id: None, } } @@ -227,6 +242,7 @@ impl ListenerConfig { l4: ServerAddress::Uds(addr.into(), None), tls: None, l4_buffer: L4BufferSettings::default(), + fd_transfer_id: None, } } @@ -264,6 +280,16 @@ impl ListenerConfig { self } + /// Set an fd-transfer ID to distinguish listeners sharing a configured address, + /// such as repeated port 0 or `SO_REUSEPORT` listeners. + /// + /// Keep the ID stable across upgrades. Mixed-version upgrades may rebind the listener. + /// IDs must be non-empty without ASCII whitespace, or listener binding fails. + pub fn fd_transfer_id(mut self, id: impl Into) -> Self { + self.fd_transfer_id = Some(id.into()); + self + } + /// Set TLS settings for this endpoint. pub fn tls(mut self, settings: TlsSettings) -> Self { self.tls = Some(settings); @@ -434,6 +460,13 @@ impl Listeners { .collect() } + pub(crate) fn fd_transfer_keys(&self) -> Vec { + self.stacks + .iter() + .map(|stack| stack.l4.fd_transfer_key(stack.fd_transfer_id.as_deref())) + .collect() + } + /// Set a connection filter for all endpoints in this listener collection #[cfg(feature = "connection_filter")] pub fn set_connection_filter(&mut self, filter: Arc) { @@ -450,11 +483,17 @@ impl Listeners { /// Add the given listener endpoint to `self`. pub fn add_listener(&mut self, endpoint: ListenerConfig) { - let ListenerConfig { l4, tls, l4_buffer } = endpoint; + let ListenerConfig { + l4, + tls, + l4_buffer, + fd_transfer_id, + } = endpoint; self.stacks.push(TransportStackBuilder { l4, tls, l4_buffer, + fd_transfer_id, #[cfg(feature = "connection_filter")] connection_filter: self.connection_filter.clone(), pre_tls_callback: self.pre_tls_callback.clone(), @@ -496,6 +535,7 @@ impl Listeners { l4, tls, l4_buffer: L4BufferSettings::default(), + fd_transfer_id: None, #[cfg(feature = "connection_filter")] connection_filter: self.connection_filter.clone(), pre_tls_callback: self.pre_tls_callback.clone(), @@ -749,6 +789,152 @@ mod test { server.await.unwrap(); } + #[cfg(unix)] + fn port_zero_listener(id: &str) -> Listeners { + let mut listeners = Listeners::new(); + listeners.add_listener(ListenerConfig::tcp("127.0.0.1:0").fd_transfer_id(id)); + listeners + } + + #[cfg(target_os = "linux")] + async fn assert_tags( + first: TransportStack, + first_addr: std::net::SocketAddr, + second: TransportStack, + second_addr: std::net::SocketAddr, + ) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let first_server = tokio::spawn(async move { + let mut stream = first.accept().await.unwrap().handshake().await.unwrap(); + stream.write_all(b"a").await.unwrap(); + stream.flush().await.unwrap(); + }); + let second_server = tokio::spawn(async move { + let mut stream = second.accept().await.unwrap().handshake().await.unwrap(); + stream.write_all(b"b").await.unwrap(); + stream.flush().await.unwrap(); + }); + let mut first_client = TcpStream::connect(first_addr).await.unwrap(); + let mut second_client = TcpStream::connect(second_addr).await.unwrap(); + let mut first_tag = [0]; + let mut second_tag = [0]; + first_client.read_exact(&mut first_tag).await.unwrap(); + second_client.read_exact(&mut second_tag).await.unwrap(); + assert_eq!(first_tag, *b"a"); + assert_eq!(second_tag, *b"b"); + first_server.await.unwrap(); + second_server.await.unwrap(); + } + + #[cfg(unix)] + #[test] + fn fd_transfer_keys_preserve_legacy_addresses() { + let mut listeners = Listeners::new(); + listeners.add_tcp("127.0.0.1:8080"); + listeners.add_uds("/tmp/pingora-transfer-key.sock", None); + listeners.add_tcp("127.0.0.1:0"); + listeners.add_listener(ListenerConfig::tcp("127.0.0.1:0").fd_transfer_id("proxy")); + + assert_eq!( + listeners.fd_transfer_keys(), + [ + "127.0.0.1:8080", + "/tmp/pingora-transfer-key.sock", + "127.0.0.1:0", + "127.0.0.1:0#id=proxy", + ] + ); + } + + #[tokio::test] + async fn whitespace_fd_transfer_id_fails_to_build() { + let mut listeners = Listeners::new(); + listeners.add_listener(ListenerConfig::tcp("127.0.0.1:0").fd_transfer_id("not valid")); + + let error = match listeners + .build( + #[cfg(unix)] + None, + ) + .await + { + Ok(_) => panic!("invalid transfer ID unexpectedly succeeded"), + Err(error) => error, + }; + assert_eq!(error.etype(), &BindError); + } + + #[cfg(unix)] + #[tokio::test] + async fn duplicate_fd_transfer_ids_fail() { + use crate::server::{Fds, ListenFds}; + use parking_lot::Mutex; + + let fds: ListenFds = Arc::new(Mutex::new(Fds::new())); + let mut first = port_zero_listener("duplicate"); + let mut second = port_zero_listener("duplicate"); + let _first = first.build(Some(fds.clone())).await.unwrap(); + let error = match second.build(Some(fds)).await { + Ok(_) => panic!("duplicate transfer ID unexpectedly succeeded"), + Err(error) => error, + }; + assert!(error + .to_string() + .contains("duplicate listener transfer identity")); + } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn stable_ids_survive_fd_transfer() { + use crate::server::{Fds, ListenFds}; + use parking_lot::Mutex; + + let fds: ListenFds = Arc::new(Mutex::new(Fds::new())); + let mut first = port_zero_listener("first"); + let mut second = port_zero_listener("second"); + let first_key = first.fd_transfer_keys().remove(0); + let second_key = second.fd_transfer_keys().remove(0); + let first_endpoints = first.build(Some(fds.clone())).await.unwrap(); + let second_endpoints = second.build(Some(fds.clone())).await.unwrap(); + let first_addr = first_endpoints[0].l4.local_addr().unwrap(); + let second_addr = second_endpoints[0].l4.local_addr().unwrap(); + assert_ne!(first_addr, second_addr); + assert_tags( + first_endpoints[0].clone(), + first_addr, + second_endpoints[0].clone(), + second_addr, + ) + .await; + + let path = format!("/tmp/pingora-port-zero-{}.sock", rand::random::()); + let receiver_path = path.clone(); + let receiver = std::thread::spawn(move || { + let mut received = Fds::new(); + received.get_from_sock(receiver_path.as_str()).unwrap(); + received + }); + fds.lock().send_to_sock(path.as_str()).unwrap(); + let received: ListenFds = Arc::new(Mutex::new(receiver.join().unwrap())); + assert!(received.lock().get(&first_key).is_some()); + assert!(received.lock().get(&second_key).is_some()); + + let mut first_after = port_zero_listener("first") + .build(Some(received.clone())) + .await + .unwrap(); + let mut second_after = port_zero_listener("second") + .build(Some(received)) + .await + .unwrap(); + let first_after = first_after.pop().unwrap(); + let second_after = second_after.pop().unwrap(); + assert_eq!(first_after.l4.local_addr().unwrap(), first_addr); + assert_eq!(second_after.l4.local_addr().unwrap(), second_addr); + assert_tags(first_after, first_addr, second_after, second_addr).await; + } + #[cfg(feature = "connection_filter")] #[test] fn test_connection_filter_inheritance() { diff --git a/pingora-core/src/server/transfer_fd/mod.rs b/pingora-core/src/server/transfer_fd/mod.rs index 9a2e36933..229599f0a 100644 --- a/pingora-core/src/server/transfer_fd/mod.rs +++ b/pingora-core/src/server/transfer_fd/mod.rs @@ -35,19 +35,38 @@ use std::{thread, time}; /// Container for open file descriptors and their associated bind addresses. pub struct Fds { map: HashMap, + local: HashSet, } impl Fds { pub fn new() -> Self { Fds { map: HashMap::new(), + local: HashSet::new(), } } pub fn add(&mut self, bind: String, fd: RawFd) { + self.local.insert(bind.clone()); self.map.insert(bind, fd); } + pub(crate) fn try_add(&mut self, bind: String, fd: RawFd) -> Result<(), Error> { + if self.map.contains_key(&bind) { + return Err(Errno::EEXIST); + } + self.add(bind, fd); + Ok(()) + } + + pub(crate) fn is_local(&self, bind: &str) -> bool { + self.local.contains(bind) + } + + pub(crate) fn mark_local(&mut self, bind: &str) { + self.local.insert(bind.to_string()); + } + pub fn get(&self, bind: &str) -> Option<&RawFd> { self.map.get(bind) } @@ -63,6 +82,7 @@ impl Fds { pub fn deserialize(&mut self, binds: Vec, fds: Vec) { assert_eq!(binds.len(), fds.len()); for (bind, fd) in binds.into_iter().zip(fds) { + self.local.remove(&bind); self.map.insert(bind, fd); } } @@ -108,6 +128,7 @@ impl Fds { closed.push(bind.clone()); false }); + self.local.retain(|bind| keep.contains(bind)); closed } } diff --git a/pingora-core/src/services/listening.rs b/pingora-core/src/services/listening.rs index 016ed4ede..bbab1599f 100644 --- a/pingora-core/src/services/listening.rs +++ b/pingora-core/src/services/listening.rs @@ -334,6 +334,6 @@ impl ServiceTrait for Service { } fn listen_addresses(&self) -> Option> { - Some(self.listeners.addresses()) + Some(self.listeners.fd_transfer_keys()) } } From a26c6db30720d94b850a0fd1eebd8ff672432f61 Mon Sep 17 00:00:00 2001 From: torinnd <81178324+torinnd@users.noreply.github.com> Date: Tue, 1 Sep 2026 14:31:17 +0100 Subject: [PATCH 2/2] services: report listener addresses after binding Allow callers to subscribe to the actual addresses of a listening service before moving it into Server. The service publishes every bound TCP or Unix address after all listeners bind and before accept loops start. This makes full-server tests with port 0 race-free while Pingora retains ownership of listening fds. TCP port-0 listeners can use stable transfer IDs from the parent change to remain distinct and retain their ports across new-to-new graceful upgrades. The transfer-ID API documents that explicitly identified listeners may rebind during a mixed-version upgrade or rollback. Tests connect through published addresses and run two services with the same 127.0.0.1:0 configuration and distinct transfer IDs. --- pingora-core/src/listeners/l4.rs | 20 ++- pingora-core/src/listeners/mod.rs | 30 +++- pingora-core/src/protocols/l4/listener.rs | 12 +- pingora-core/src/services/listening.rs | 26 +++ .../listening_service_bound_addresses.rs | 155 ++++++++++++++++++ 5 files changed, 222 insertions(+), 21 deletions(-) create mode 100644 pingora-core/tests/listening_service_bound_addresses.rs diff --git a/pingora-core/src/listeners/l4.rs b/pingora-core/src/listeners/l4.rs index bf8bb9287..80ff896f3 100644 --- a/pingora-core/src/listeners/l4.rs +++ b/pingora-core/src/listeners/l4.rs @@ -455,12 +455,7 @@ impl ListenerEndpoint { self.listen_addr.as_ref() } - /// Return the local address this endpoint is bound to. - /// - /// Useful when the listener was bound to port 0 (OS-assigned) to - /// discover the actual port. - #[cfg(test)] - pub fn local_addr(&self) -> Option { + pub(crate) fn local_addr(&self) -> std::io::Result { self.listener.local_addr() } @@ -558,7 +553,7 @@ mod test { #[cfg(windows)] let listener = builder.listen().await.unwrap(); - let addr = listener.local_addr().unwrap(); + let addr = *listener.local_addr().unwrap().as_inet().unwrap(); tokio::spawn(async move { // just try to accept once @@ -586,7 +581,7 @@ mod test { #[cfg(windows)] let listener = builder.listen().await.unwrap(); - let port = listener.local_addr().unwrap().port(); + let port = listener.local_addr().unwrap().as_inet().unwrap().port(); tokio::spawn(async move { // just try to accept twice @@ -611,6 +606,15 @@ mod test { builder.listen_addr(ServerAddress::Uds(addr.into(), None)); let listener = builder.listen(None).await.unwrap(); + assert_eq!( + listener + .local_addr() + .unwrap() + .as_unix() + .unwrap() + .as_pathname(), + Some(std::path::Path::new(addr)) + ); tokio::spawn(async move { // just try to accept once diff --git a/pingora-core/src/listeners/mod.rs b/pingora-core/src/listeners/mod.rs index c3e9531c7..d4414a1e2 100644 --- a/pingora-core/src/listeners/mod.rs +++ b/pingora-core/src/listeners/mod.rs @@ -316,6 +316,10 @@ impl TransportStack { self.l4.as_str() } + pub fn local_addr(&self) -> std::io::Result { + self.l4.local_addr() + } + pub async fn accept(&self) -> Result { let stream = self.l4.accept().await?; Ok(UninitializedStream { @@ -592,7 +596,7 @@ mod test { assert_eq!(listeners.len(), 2); let addrs: Vec<_> = listeners .iter() - .map(|s| s.l4.local_addr().unwrap()) + .map(|s| *s.local_addr().unwrap().as_inet().unwrap()) .collect(); for listener in listeners { tokio::spawn(async move { @@ -897,8 +901,18 @@ mod test { let second_key = second.fd_transfer_keys().remove(0); let first_endpoints = first.build(Some(fds.clone())).await.unwrap(); let second_endpoints = second.build(Some(fds.clone())).await.unwrap(); - let first_addr = first_endpoints[0].l4.local_addr().unwrap(); - let second_addr = second_endpoints[0].l4.local_addr().unwrap(); + let first_addr = *first_endpoints[0] + .l4 + .local_addr() + .unwrap() + .as_inet() + .unwrap(); + let second_addr = *second_endpoints[0] + .l4 + .local_addr() + .unwrap() + .as_inet() + .unwrap(); assert_ne!(first_addr, second_addr); assert_tags( first_endpoints[0].clone(), @@ -930,8 +944,14 @@ mod test { .unwrap(); let first_after = first_after.pop().unwrap(); let second_after = second_after.pop().unwrap(); - assert_eq!(first_after.l4.local_addr().unwrap(), first_addr); - assert_eq!(second_after.l4.local_addr().unwrap(), second_addr); + assert_eq!( + first_after.l4.local_addr().unwrap().as_inet(), + Some(&first_addr) + ); + assert_eq!( + second_after.l4.local_addr().unwrap().as_inet(), + Some(&second_addr) + ); assert_tags(first_after, first_addr, second_after, second_addr).await; } diff --git a/pingora-core/src/protocols/l4/listener.rs b/pingora-core/src/protocols/l4/listener.rs index a6055267a..f96268426 100644 --- a/pingora-core/src/protocols/l4/listener.rs +++ b/pingora-core/src/protocols/l4/listener.rs @@ -24,6 +24,7 @@ use tokio::net::TcpListener; use tokio::net::UnixListener; use crate::protocols::digest::{GetSocketDigest, SocketDigest}; +use crate::protocols::l4::socket::SocketAddr; use crate::protocols::l4::stream::Stream; /// The type for generic listener for both TCP and Unix domain socket @@ -68,16 +69,11 @@ impl AsRawSocket for Listener { impl Listener { /// Return the local address this listener is bound to. - /// - /// For TCP listeners this is the resolved address (including the - /// OS-assigned port when the listener was bound to port 0). - /// Returns `None` for non-TCP listeners (e.g. Unix domain sockets). - #[cfg(test)] - pub fn local_addr(&self) -> Option { + pub(crate) fn local_addr(&self) -> io::Result { match self { - Self::Tcp(l) => l.local_addr().ok(), + Self::Tcp(listener) => listener.local_addr().map(Into::into), #[cfg(unix)] - Self::Unix(_) => None, + Self::Unix(listener) => listener.local_addr()?.try_into().map_err(io::Error::other), } } diff --git a/pingora-core/src/services/listening.rs b/pingora-core/src/services/listening.rs index bbab1599f..0420e33af 100644 --- a/pingora-core/src/services/listening.rs +++ b/pingora-core/src/services/listening.rs @@ -25,6 +25,7 @@ use crate::listeners::AcceptAllFilter; use crate::listeners::{ ConnectionFilter, ListenerConfig, Listeners, ServerAddress, TcpSocketOptions, TransportStack, }; +use crate::protocols::l4::socket::SocketAddr; use crate::protocols::Stream; #[cfg(unix)] use crate::server::ListenFds; @@ -39,10 +40,14 @@ use pingora_timeout::timeout; use std::fs::Permissions; use std::sync::Arc; use std::time::Duration; +use tokio::sync::watch; /// Override the runtime options used to run a listening service. pub type RuntimeOptsOverride = Arc Option + Send + Sync>; +/// A receiver for the addresses of a service's bound listeners. +pub type BoundAddressWatch = watch::Receiver>>; + /// The type of service that is associated with a list of listening endpoints and a particular application pub struct Service { name: String, @@ -51,6 +56,7 @@ pub struct Service { /// The number of preferred threads. `None` to follow global setting. pub threads: Option, runtime_opts_override: Option, + bound_addresses: watch::Sender>>, #[cfg(feature = "connection_filter")] connection_filter: Arc, } @@ -64,6 +70,7 @@ impl Service { app_logic: Some(app_logic), threads: None, runtime_opts_override: None, + bound_addresses: watch::channel(None).0, #[cfg(feature = "connection_filter")] connection_filter: Arc::new(AcceptAllFilter), } @@ -78,6 +85,7 @@ impl Service { app_logic: Some(app_logic), threads: None, runtime_opts_override: None, + bound_addresses: watch::channel(None).0, #[cfg(feature = "connection_filter")] connection_filter: Arc::new(AcceptAllFilter), } @@ -129,6 +137,17 @@ impl Service { &mut self.listeners } + /// Subscribe to the addresses after all of this service's listeners bind. + /// + /// The addresses are published once, in endpoint insertion order, before + /// accept loops start. The watch closes without a value if startup fails. + /// + /// During graceful upgrade, a TCP port-0 endpoint with a stable + /// [`ListenerConfig::fd_transfer_id`] retains its assigned port. + pub fn watch_bound_addresses(&self) -> BoundAddressWatch { + self.bound_addresses.subscribe() + } + // the follow add* function has no effect if the server is already started /// Add a TCP listening endpoint with the given address (e.g., `127.0.0.1:8000`). @@ -292,6 +311,13 @@ impl ServiceTrait for Service { .await .expect("Failed to build listeners"); + let bound_addresses = endpoints + .iter() + .map(TransportStack::local_addr) + .collect::>>() + .expect("Failed to get bound listener addresses"); + self.bound_addresses.send_replace(Some(bound_addresses)); + let app_logic = self .app_logic .take() diff --git a/pingora-core/tests/listening_service_bound_addresses.rs b/pingora-core/tests/listening_service_bound_addresses.rs new file mode 100644 index 000000000..738f0b4d2 --- /dev/null +++ b/pingora-core/tests/listening_service_bound_addresses.rs @@ -0,0 +1,155 @@ +// Copyright 2026 Cloudflare, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#![cfg(unix)] + +use async_trait::async_trait; +use pingora_core::apps::ServerApp; +use pingora_core::listeners::ListenerConfig; +use pingora_core::protocols::Stream; +use pingora_core::server::configuration::ServerConf; +use pingora_core::server::{RunArgs, Server, ShutdownSignal, ShutdownSignalWatch, ShutdownWatch}; +use pingora_core::services::listening::{BoundAddressWatch, Service as ListeningService}; +use std::sync::Arc; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::Notify; + +struct TaggedApp(u8); + +#[async_trait] +impl ServerApp for TaggedApp { + async fn process_new( + self: &Arc, + mut stream: Stream, + _shutdown: &ShutdownWatch, + ) -> Option { + let mut byte = [0]; + stream.read_exact(&mut byte).await.unwrap(); + stream.write_all(&[self.0]).await.unwrap(); + None + } +} + +struct TestShutdown(Arc); + +#[async_trait] +impl ShutdownSignalWatch for TestShutdown { + async fn recv(&self) -> ShutdownSignal { + self.0.notified().await; + ShutdownSignal::FastShutdown + } +} + +struct RunningServer { + shutdown: Arc, + thread: Option>, +} + +impl RunningServer { + fn start(mut server: Server) -> Self { + let shutdown = Arc::new(Notify::new()); + let shutdown_watch = TestShutdown(shutdown.clone()); + let thread = std::thread::spawn(move || { + server.bootstrap(); + server.run(RunArgs { + shutdown_signal: Box::new(shutdown_watch), + }); + }); + Self { + shutdown, + thread: Some(thread), + } + } +} + +impl Drop for RunningServer { + fn drop(&mut self) { + self.shutdown.notify_one(); + if let Some(thread) = self.thread.take() { + let result = thread.join(); + if !std::thread::panicking() { + result.unwrap(); + } + } + } +} + +async fn bound_tcp_address(watch: &mut BoundAddressWatch) -> std::net::SocketAddr { + let addresses = tokio::time::timeout( + std::time::Duration::from_secs(10), + watch.wait_for(Option::is_some), + ) + .await + .expect("listener did not bind in time") + .expect("listening service exited before binding") + .clone() + .unwrap(); + *addresses[0].as_inet().unwrap() +} + +async fn assert_service(address: std::net::SocketAddr, expected: u8) { + let mut stream = tokio::net::TcpStream::connect(address).await.unwrap(); + stream.write_all(b"x").await.unwrap(); + let mut response = [0]; + stream.read_exact(&mut response).await.unwrap(); + assert_eq!(response, [expected]); +} + +#[test] +fn reports_os_assigned_address() { + let mut server = Server::new_with_opt_and_conf(None, ServerConf::default()); + let mut service = ListeningService::new("listener".to_string(), TaggedApp(b'1')); + service.add_tcp("127.0.0.1:0"); + let mut bound_addresses = service.watch_bound_addresses(); + server.add_service(service); + let _server = RunningServer::start(server); + + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let address = bound_tcp_address(&mut bound_addresses).await; + assert_ne!(address.port(), 0); + assert_service(address, b'1').await; + }); +} + +#[test] +fn duplicate_port_zero_listeners_bind_distinct_sockets() { + let mut server = Server::new_with_opt_and_conf(None, ServerConf::default()); + + let mut first = ListeningService::new("first".to_string(), TaggedApp(b'1')); + first.add_listener(ListenerConfig::tcp("127.0.0.1:0").fd_transfer_id("first")); + let mut first_addresses = first.watch_bound_addresses(); + server.add_service(first); + + let mut second = ListeningService::new("second".to_string(), TaggedApp(b'2')); + second.add_listener(ListenerConfig::tcp("127.0.0.1:0").fd_transfer_id("second")); + let mut second_addresses = second.watch_bound_addresses(); + server.add_service(second); + + let _server = RunningServer::start(server); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let first = bound_tcp_address(&mut first_addresses).await; + let second = bound_tcp_address(&mut second_addresses).await; + assert_ne!(first, second); + assert_service(first, b'1').await; + assert_service(second, b'2').await; + }); +}