Merge branch 'macos-handle-closed-routing-socket' into main

2 files changed, 32 insertions, 7 deletions

diff --git a/talpid-routing/src/unix/macos/routing_socket.rs b/talpid-routing/src/unix/macos/routing_socket.rs
index ab02f1cff2..763dbb9264 100644
--- a/talpid-routing/src/unix/macos/routing_socket.rs
+++ b/talpid-routing/src/unix/macos/routing_socket.rs
@@ -34,6 +34,23 @@ pub enum Error {
     ResponseTimeout,
 }
 
+impl Error {
+    /// Return the underlying `io::Error` (or `None`)
+    pub fn as_io_error(&self) -> Option<&io::Error> {
+        use std::error::Error;
+        self.source()
+            .and_then(|source| source.downcast_ref::<io::Error>())
+    }
+
+    /// Return whether an operation failed because the socket has been shut down
+    pub fn is_shutdown(&self) -> bool {
+        // ENOTCONN is returned when the socket is shut down (e.g., due to `pid_shutdown_sockets`)
+        self.as_io_error()
+            .map(|io_error| io_error.kind() == io::ErrorKind::NotConnected)
+            .unwrap_or(false)
+    }
+}
+
 type Result<T> = std::result::Result<T, Error>;
 
 const RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);
@@ -61,7 +78,7 @@ impl RoutingSocket {
 
     pub async fn recv_msg(&mut self, mut buf: &mut [u8]) -> Result<usize> {
         if let Some(buffered_msg) = self.buf.pop_front() {
-            let bytes_written = buf.write(&buffered_msg).map_err(Error::Write)?;
+            let bytes_written = buf.write(&buffered_msg).map_err(Error::Read)?;
             return Ok(bytes_written);
         }
         self.read_next_msg(buf).await
@@ -85,7 +102,7 @@ impl RoutingSocket {
         }
     }
 
-    pub async fn wait_for_response(&mut self, response_num: i32) -> Result<Vec<u8>> {
+    async fn wait_for_response(&mut self, response_num: i32) -> Result<Vec<u8>> {
         loop {
             talpid_types::detect_flood!();
 
diff --git a/talpid-routing/src/unix/macos/watch.rs b/talpid-routing/src/unix/macos/watch.rs
index 7c0f68ccd0..70271a85d4 100644
--- a/talpid-routing/src/unix/macos/watch.rs
+++ b/talpid-routing/src/unix/macos/watch.rs
@@ -46,11 +46,19 @@ impl RoutingTable {
 
     pub async fn next_message(&mut self) -> Result<RouteSocketMessage> {
         let mut buf = [0u8; 2048];
-        let bytes_read = self
-            .socket
-            .recv_msg(&mut buf)
-            .await
-            .map_err(Error::RoutingSocket)?;
+
+        let bytes_read = loop {
+            match self.socket.recv_msg(&mut buf).await {
+                Ok(bytes_read) => break bytes_read,
+                Err(error) if error.is_shutdown() => {
+                    log::debug!("Recreating shut down socket");
+                    self.socket =
+                        routing_socket::RoutingSocket::new().map_err(Error::RoutingSocket)?;
+                }
+                Err(error) => return Err(Error::RoutingSocket(error)),
+            }
+        };
+
         let msg_buf = &buf[0..bytes_read];
         data::RouteSocketMessage::parse_message(msg_buf).map_err(Error::InvalidMessage)
     }