ops: Classify std::io::ErrorKind::BrokenPipe as a retryable transient error

Signed-off-by: Joshua Potts <[email protected]>

Author: iamjpotts

src/execute.rs

@@ -41,10 +41,7 @@ use tonic::transport::Channel;
 use triomphe::Arc;
 
 use crate::client::NetworkData;
-use crate::execute::error::{
-    is_hyper_canceled,
-    is_tonic_status_transient,
-};
+use crate::execute::error::is_tonic_status_transient;
 use crate::ping_query::PingQuery;
 use crate::{
     client,
@@ -334,13 +331,6 @@ fn map_tonic_error(
             retry::Error::Transient(status.into())
         }
 
-        // if the proxy cancels the request (IE it's `Unavailable`/`ResourceExausted`) treat it like a transient error.
-        tonic::Code::Unknown if is_hyper_canceled(&status) => {
-            network.mark_node_unhealthy(node_index);
-
-            retry::Error::Transient(status.into())
-        }
-
         // todo: find a way to make this less fragile
         // hack:
         // if this happens:
@@ -360,7 +350,9 @@ fn map_tonic_error(
             }
         }
 
-        tonic::Code::Internal if is_tonic_status_transient(&status) => {
+        _ if is_tonic_status_transient(&status) => {
+            network.mark_node_unhealthy(node_index);
+
             retry::Error::Transient(status.into())
         }
 

src/execute/error.rs

@@ -21,8 +21,56 @@ use std::error::Error;
 
 use serde::de::StdError;
 
-/// Punches through all the layers of `tonic::Status` sources to check if this is a `hyper::Error` that is canceled.
-pub(super) fn is_hyper_canceled(status: &tonic::Status) -> bool {
+fn has_transient_io_error<E: StdError>(error: E) -> bool {
+    let Some(source) = error.source() else {
+        return false;
+    };
+
+    if let Some(io_error) = source.downcast_ref::<std::io::Error>() {
+        is_io_error_transient(io_error)
+    } else {
+        false
+    }
+}
+
+// tonic 0.11 (current dependency)
+fn is_hyper_0_error_transient(error: &hyper_0::Error) -> bool {
+    if error.is_canceled() || has_transient_io_error(error) {
+        true
+    } else if let Some(source) = error.source() {
+        if let Some(h2_error) = source.downcast_ref::<h2_03::Error>() {
+            h2_error.is_go_away()
+        } else {
+            false
+        }
+    } else {
+        false
+    }
+}
+
+// tonic 0.12
+fn is_hyper_error_transient(error: &hyper::Error) -> bool {
+    if error.is_canceled() || has_transient_io_error(error) {
+        true
+    } else if let Some(source) = error.source() {
+        if let Some(h2_error) = source.downcast_ref::<h2::Error>() {
+            h2_error.is_go_away()
+        } else {
+            false
+        }
+    } else {
+        false
+    }
+}
+
+fn is_io_error_transient(error: &std::io::Error) -> bool {
+    match error.kind() {
+        std::io::ErrorKind::BrokenPipe => true,
+        _ => false,
+    }
+}
+
+pub(super) fn is_tonic_status_transient(status: &tonic::Status) -> bool {
     let source = status
         .source()
         .and_then(|it| it.downcast_ref::<tonic::transport::Error>())
@@ -33,11 +81,9 @@ pub(super) fn is_hyper_canceled(status: &tonic::Status) -> bool {
     };
 
     if let Some(hyper_0) = source.downcast_ref::<hyper_0::Error>() {
-        // tonic 0.11 (current dependency)
-        hyper_0.is_canceled()
-    } else if let Some(hyper_1) = source.downcast_ref::<hyper::Error>() {
-        // tonic 0.12
-        hyper_1.is_canceled()
+        is_hyper_0_error_transient(hyper_0)
+    } else if let Some(hyper) = source.downcast_ref::<hyper::Error>() {
+        is_hyper_error_transient(hyper)
     } else {
         false
     }
@@ -48,47 +94,22 @@ pub(super) fn is_hyper_canceled(status: &tonic::Status) -> bool {
 /// Because hyper does not expose constructors for its error variants, there is no
 /// reasonable way to construct a test for positive detection of a hyper cancellation.
 #[cfg(test)]
-mod test_is_hyper_canceled {
+mod test_is_tonic_status_transient {
     use tonic::Code;
 
-    use super::is_hyper_canceled;
+    use super::is_tonic_status_transient;
 
     #[test]
     fn ignores_tonic_abort() {
         let input = tonic::Status::new(Code::Aborted, "foo");
 
-        assert!(!is_hyper_canceled(&input));
+        assert!(!is_tonic_status_transient(&input));
     }
 
     #[test]
     fn ignores_tonic_cancel() {
         let input = tonic::Status::new(Code::Cancelled, "foo");
 
-        assert!(!is_hyper_canceled(&input));
-    }
-}
-
-pub(super) fn is_tonic_status_transient(status: &tonic::Status) -> bool {
-    let source = status
-        .source()
-        .and_then(|it| it.downcast_ref::<tonic::transport::Error>())
-        .and_then(StdError::source);
-
-    let Some(source) = source else {
-        return false;
-    };
-
-    if let Some(hyper_0) = source.downcast_ref::<hyper_0::Error>() {
-        // tonic 0.11 (current dependency)
-        let source: Option<&h2_03::Error> = hyper_0.source().and_then(|s| s.downcast_ref());
-
-        source.map(|s| s.is_go_away()).unwrap_or_default()
-    } else if let Some(hyper_1) = source.downcast_ref::<hyper::Error>() {
-        // tonic 0.12
-        let source: Option<&h2::Error> = hyper_1.source().and_then(|s| s.downcast_ref());
-
-        source.map(|s| s.is_go_away()).unwrap_or_default()
-    } else {
-        false
+        assert!(!is_tonic_status_transient(&input));
     }
 }