From a24663438c67b9b10159679d639f9ba89fa71b57 Mon Sep 17 00:00:00 2001 From: Wael Jami Date: Thu, 2 Jul 2026 14:19:32 +0200 Subject: [PATCH] fix: make serve_listens idempotent to avoid duplicate bind AddrInUse TcpListenerConnection (and the TLS/WebSocket listeners) bind their socket inside serve_listener() on every call, not in new(). Endpoint::serve() already calls serve_listens() internally, so an application that also calls serve_listens() explicitly (e.g. to have listeners ready before serve() is polled) triggers a second bind on the same address, which fails with AddrInUse and logs a spurious 'Failed to serve listener' warning. SO_REUSEADDR does not allow two live listeners on the same (ip, port). Track already-served listener addresses in TransportLayerInner and make serve_listener a no-op for them. Failed serves are removed from the set so they can be retried, and del_listener clears the address so a re-added listener is served again. --- src/transport/transport_layer.rs | 60 ++++++++++++++++++++++++++++++-- 1 file changed, 57 insertions(+), 3 deletions(-) diff --git a/src/transport/transport_layer.rs b/src/transport/transport_layer.rs index 406febe0..996fec1c 100644 --- a/src/transport/transport_layer.rs +++ b/src/transport/transport_layer.rs @@ -8,7 +8,7 @@ use crate::transport::connection::TransportReceiver; use crate::{transport::TransportEvent, Result}; use async_trait::async_trait; use parking_lot::{Mutex, RwLock}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::net::IpAddr; use std::sync::Arc; use tokio::select; @@ -100,6 +100,7 @@ impl DomainResolver for DefaultDomainResolver { pub struct TransportLayerInner { pub(crate) cancel_token: CancellationToken, listens: Arc>>, // listening transports + served_listens: Mutex>, // addresses already being served connections: Arc>>, // outbound/inbound connections pub(crate) transport_tx: TransportSender, pub(crate) transport_rx: Mutex>, @@ -124,6 +125,7 @@ impl TransportLayer { let inner = TransportLayerInner { cancel_token, listens: Arc::new(RwLock::new(Vec::new())), + served_listens: Mutex::new(HashSet::new()), connections: Arc::new(RwLock::new(HashMap::new())), transport_tx, transport_rx: Mutex::new(Some(transport_rx)), @@ -166,6 +168,10 @@ impl TransportLayer { self.inner.lookup(target, self.outbound.as_ref(), key).await } + /// Serve all registered listeners. Idempotent: listeners whose address is + /// already being served are skipped, so calling this explicitly before + /// `Endpoint::serve()` (which also calls it) does not attempt a duplicate + /// bind that would fail with AddrInUse. pub async fn serve_listens(&self) -> Result<()> { let listens = self.inner.listens.read().clone(); for transport in listens { @@ -246,6 +252,7 @@ impl TransportLayerInner { pub(super) fn del_listener(&self, addr: &SipAddr) { self.listens.write().retain(|t| t.get_addr() != addr); + self.served_listens.lock().remove(addr); } pub(super) fn add_connection(&self, connection: SipConnection) { @@ -354,8 +361,13 @@ impl TransportLayerInner { } pub(super) async fn serve_listener(self: Arc, transport: SipConnection) -> Result<()> { + let addr = transport.get_addr().clone(); + if !self.served_listens.lock().insert(addr.clone()) { + debug!(%addr, "listener already served, skipping"); + return Ok(()); + } let sender = self.transport_tx.clone(); - match transport { + let result = match transport { SipConnection::Udp(transport) => { let transport_layer_inner = self.clone(); tokio::spawn(async move { @@ -380,7 +392,11 @@ impl TransportLayerInner { ); Ok(()) } + }; + if result.is_err() { + self.served_listens.lock().remove(&addr); } + result } pub fn serve_connection(&self, transport: SipConnection) { @@ -428,7 +444,7 @@ mod tests { use crate::sip::uri::ParamsExt; use crate::sip::{Host, HostWithPort, Transport}; use crate::{ - transport::{udp::UdpConnection, SipAddr}, + transport::{tcp_listener::TcpListenerConnection, udp::UdpConnection, SipAddr}, Result, }; @@ -553,4 +569,42 @@ mod tests { Ok(()) } + + #[tokio::test] + async fn test_serve_listener_is_idempotent() -> Result<()> { + let tl = super::TransportLayer::new(tokio_util::sync::CancellationToken::new()); + + // Reserve a concrete port so a duplicate bind on it would fail + let probe = std::net::TcpListener::bind("127.0.0.1:0")?; + let local_addr = probe.local_addr()?; + drop(probe); + + let listener = TcpListenerConnection::new( + SipAddr { + r#type: Some(Transport::Tcp), + addr: local_addr.into(), + }, + None, + ) + .await?; + let connection: super::SipConnection = listener.into(); + tl.add_transport(connection.clone()); + + // First serve binds the listener + super::TransportLayerInner::serve_listener(tl.inner.clone(), connection.clone()).await?; + + // Serving the same listener again must be a no-op instead of a + // second bind that fails with AddrInUse + super::TransportLayerInner::serve_listener(tl.inner.clone(), connection.clone()).await?; + + // The listener must still accept connections + tokio::net::TcpStream::connect(local_addr).await?; + + // Removing the listener frees its address for a future serve + tl.del_transport(connection.get_addr()); + assert!(tl.inner.served_listens.lock().is_empty()); + + drop(tl); + Ok(()) + } }