swbus: retry listener bind failures

Cargo.toml

@@ -37,7 +37,7 @@ unused_import_braces = 'warn'
 # Async framework
 tokio = { version = "1.37", features = ["macros", "net", "rt-multi-thread", "signal"] }
 tokio-util = { version = "0.7", features = ["rt"] }
-tokio-stream = "0.1"
+tokio-stream = { version = "0.1", features = ["net"] }
 tower = { version = "0.4", features = ["util"] }
 
 # Log and error handling

crates/swbus-core/src/mux/service.rs

@@ -7,20 +7,48 @@ use std::io;
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::time::Duration;
 use swbus_config::SwbusConfig;
 use swbus_proto::result::*;
 use swbus_proto::swbus::swbus_service_server::{SwbusService, SwbusServiceServer};
 use swbus_proto::swbus::*;
+use tokio::net::TcpListener;
 use tokio::sync::{
     mpsc,
     oneshot::{self, Receiver, Sender},
 };
-use tokio_stream::wrappers::ReceiverStream;
+use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
 use tokio_stream::Stream;
 use tokio_util::sync::CancellationToken;
 use tonic::{transport::Server, Request, Response, Status, Streaming};
 use tracing::*;
 
+const LISTENER_RETRY_INTERVAL: Duration = Duration::from_secs(1);
+const LISTENER_RETRY_TIMEOUT: Duration = Duration::from_secs(180);
+
+async fn bind_listener_with_retry(
+    addr: SocketAddr,
+    retry_interval: Duration,
+    retry_timeout: Duration,
+) -> io::Result<TcpListener> {
+    let deadline = tokio::time::Instant::now() + retry_timeout;
+
+    loop {
+        match TcpListener::bind(addr).await {
+            Ok(listener) => return Ok(listener),
+            Err(error) if tokio::time::Instant::now() < deadline => {
+                warn!(%addr, %error, "Failed to bind swbus listener; retrying");
+                let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+                tokio::time::sleep(retry_interval.min(remaining)).await;
+            }
+            Err(error) => {
+                error!(%addr, %error, timeout_seconds=retry_timeout.as_secs(), "Failed to bind swbus listener");
+                return Err(error);
+            }
+        }
+    }
+}
+
 pub struct SwbusServiceHost {
     swbus_server_addr: SocketAddr,
     mux: Option<Arc<SwbusMultiplexer>>,
@@ -66,6 +94,15 @@ impl SwbusServiceHost {
             ));
         }
 
+        let listener = bind_listener_with_retry(addr, LISTENER_RETRY_INTERVAL, LISTENER_RETRY_TIMEOUT)
+            .await
+            .map_err(|error| {
+                SwbusError::connection(
+                    SwbusErrorCode::ConnectionError,
+                    io::Error::new(error.kind(), format!("Failed to bind at {addr}: {error}")),
+                )
+            })?;
+
         // create mux and set route announce task queue
         let mut mux = SwbusMultiplexer::new(config.routes);
         let (route_annouce_task_tx, route_annouce_task_rx) = mpsc::channel::<RouteAnnounceTask>(100);
@@ -95,7 +132,7 @@ impl SwbusServiceHost {
 
         Server::builder()
             .add_service(SwbusServiceServer::new(self))
-            .serve_with_shutdown(addr, async {
+            .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async {
                 shutdown_rx.await.ok();
                 info!("SwbusServiceServer received shutdown signal");
                 conn_store_clone.shutdown().await;
@@ -104,7 +141,7 @@ impl SwbusServiceHost {
             .map_err(|e| {
                 SwbusError::connection(
                     SwbusErrorCode::ConnectionError,
-                    io::Error::other(format!("Failed to listen at {addr}: {e}")),
+                    io::Error::other(format!("Failed to serve at {addr}: {e}")),
                 )
             })?;
         debug!("SwbusServiceServer terminated");
@@ -185,3 +222,42 @@ impl SwbusService for SwbusServiceHost {
         Ok(response)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::Ipv4Addr;
+    use tokio::time::{timeout, Duration};
+
+    #[tokio::test]
+    async fn listener_bind_retries_until_address_is_available() {
+        let occupied_listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = occupied_listener.local_addr().unwrap();
+        let bind_task = tokio::spawn(bind_listener_with_retry(
+            addr,
+            Duration::from_millis(10),
+            Duration::from_secs(1),
+        ));
+
+        tokio::time::sleep(Duration::from_millis(30)).await;
+        assert!(!bind_task.is_finished());
+
+        drop(occupied_listener);
+        let listener = timeout(Duration::from_secs(1), bind_task)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        assert_eq!(listener.local_addr().unwrap(), addr);
+    }
+
+    #[tokio::test]
+    async fn listener_bind_returns_error_after_timeout() {
+        let occupied_listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
+        let addr = occupied_listener.local_addr().unwrap();
+
+        let result = bind_listener_with_retry(addr, Duration::from_millis(10), Duration::from_millis(30)).await;
+
+        assert_eq!(result.unwrap_err().kind(), io::ErrorKind::AddrInUse);
+    }
+}