fix(ecstore): classify peer RPC failures by gRPC status code (#5400)

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 use crate::cluster::rpc::client::{
-    TonicInterceptor, gen_tonic_signature_interceptor, heal_control_time_out_client, node_service_time_out_client,
-    tier_mutation_control_time_out_client,
+    TonicInterceptor, embedded_tonic_status, gen_tonic_signature_interceptor, heal_control_time_out_client,
+    is_network_like_status, message_has_network_needle, node_service_time_out_client, tier_mutation_control_time_out_client,
 };
 use crate::cluster::rpc::{set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof};
 use crate::error::{Error, Result};
@@ -471,26 +471,21 @@ impl PeerRestClient {
         self.offline.store(false, Ordering::Release);
     }
 
+    /// Whether this failure means the peer is unreachable, so it should be
+    /// gated offline and its connection evicted.
+    ///
+    /// RPC failures are classified by their typed gRPC code first
+    /// (`is_network_like_status`); an application error from a live peer must
+    /// never take it offline no matter what its message says. The substring
+    /// fallback only covers failures that exist purely as text, such as the
+    /// dial errors `get_client` wraps.
     fn is_network_like_error(err: &Error) -> bool {
-        let message = err.to_string().to_ascii_lowercase();
-        [
-            "temporarily offline",
-            "transport error",
-            "unavailable",
-            "error trying to connect",
-            "connection refused",
-            "connection reset",
-            "broken pipe",
-            "not connected",
-            "unexpected eof",
-            "timed out",
-            "deadline has elapsed",
-            "connection closed",
-            "connection aborted",
-            "tcp connect error",
-        ]
-        .iter()
-        .any(|needle| message.contains(needle))
+        if let Error::Io(io_err) = err
+            && let Some(status) = embedded_tonic_status(io_err)
+        {
+            return is_network_like_status(status);
+        }
+        message_has_network_needle(&err.to_string())
     }
 
     fn mark_offline_and_spawn_recovery(&self) {
@@ -1702,30 +1697,17 @@ fn tier_config_reload_connection_outcome(err: Error) -> TierConfigReloadOutcome
 }
 
 fn is_tier_config_reload_connection_failure(err: &Error) -> bool {
-    let message = err.to_string().to_ascii_lowercase();
+    let message = err.to_string();
+    // A bare "unavailable" is only trusted inside the local dial-failure
+    // wrapper from `get_client`, never in application text.
     if message
+        .to_ascii_lowercase()
         .split_once("can not get client, err:")
         .is_some_and(|(_, local_error)| local_error.contains("unavailable"))
     {
         return true;
     }
-    [
-        "temporarily offline",
-        "transport error",
-        "error trying to connect",
-        "connection refused",
-        "connection reset",
-        "connection closed",
-        "connection aborted",
-        "broken pipe",
-        "not connected",
-        "unexpected eof",
-        "timed out",
-        "deadline has elapsed",
-        "tcp connect error",
-    ]
-    .iter()
-    .any(|needle| message.contains(needle))
+    message_has_network_needle(&message)
 }
 
 fn tier_config_reload_remote_failure(error_info: Option<String>) -> TierConfigReloadOutcome {
@@ -2151,6 +2133,126 @@ mod tests {
         assert!(!PeerRestClient::is_network_like_error(&Error::NotImplemented));
     }
 
+    #[test]
+    fn peer_rest_client_network_classifier_uses_typed_status_code() {
+        // The one code that means "nothing is answering on this channel".
+        assert!(PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unavailable(
+            "storage layer is not initialized"
+        ))));
+        // Application statuses from a live peer must not mark it offline,
+        // even when their message contains transport-sounding words.
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::internal(
+            "failed to reload metadata for bucket \"unavailable-logs\""
+        ))));
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unauthenticated(
+            "No valid auth token"
+        ))));
+        // A request-budget expiry answered by a live peer is an application
+        // outcome, not a transport failure.
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::deadline_exceeded(
+            "heal control request expired"
+        ))));
+        // Unknown is the transport's escape hatch for a cause it could not
+        // map, and our handlers never return it, so there the text decides.
+        assert!(PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unknown(
+            "Service was not ready: transport error"
+        ))));
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unknown(
+            "peer response unknown"
+        ))));
+    }
+
+    #[test]
+    fn peer_rest_client_network_classifier_ignores_transport_words_in_application_statuses() {
+        // The reason classification reads the code rather than the text: a
+        // peer that answers is reachable, even when what it says describes a
+        // connection failure of its own. A handler interpolating a local
+        // io::Error into Status::internal, or relaying trouble with its own
+        // downstream, must not cost us the channel to a healthy peer.
+        for status in [
+            tonic::Status::internal("connection refused while dialing downstream backend"),
+            tonic::Status::internal("write failed: broken pipe"),
+            tonic::Status::unauthenticated("connection reset while validating token"),
+            tonic::Status::failed_precondition("scanner lease timed out"),
+            tonic::Status::deadline_exceeded("heal control request timed out"),
+        ] {
+            let rendered = status.to_string();
+            assert!(
+                !PeerRestClient::is_network_like_error(&Error::from(status)),
+                "an answered application status must not mark the peer offline: {rendered}"
+            );
+        }
+    }
+
+    #[test]
+    fn peer_rest_client_network_classifier_keeps_slow_peers_online() {
+        // The per-RPC channel deadline (RUSTFS_INTERNODE_RPC_TIMEOUT, 30s)
+        // surfaces as Cancelled "Timeout expired" carrying the transport
+        // cause as its source. A peer that is merely slow must stay online:
+        // gating it would spend a full recovery cycle fast-failing every RPC
+        // to a host that is still answering, turning load into a partition.
+        let timeout_status = tonic::Status::cancelled("Timeout expired");
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(timeout_status)));
+
+        let sourced = tonic::Status::from_error(Box::new(std::io::Error::other("Timeout expired")));
+        assert!(
+            std::error::Error::source(&sourced).is_some(),
+            "the transport builds this status through Status::from_error, which attaches the cause"
+        );
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(sourced)));
+    }
+
+    #[test]
+    fn rpc_status_errors_keep_their_rendering_and_hide_peer_metadata() {
+        let err = Error::from(tonic::Status::unavailable("peer gone"));
+        assert_eq!(
+            err.to_string(),
+            "Io error: code: 'The service is currently unavailable', message: \"peer gone\""
+        );
+
+        // tonic's own Debug prints the MetadataMap, i.e. every response header
+        // the peer sent; those must not reach a log through this error.
+        let mut status = tonic::Status::unavailable("peer gone");
+        status
+            .metadata_mut()
+            .insert("authorization", "Bearer secret".parse().expect("valid header value"));
+        let rendered = format!("{:?}", Error::from(status));
+        assert!(!rendered.contains("Bearer secret"), "{rendered}");
+        assert!(!rendered.contains("MetadataMap"), "{rendered}");
+    }
+
+    #[test]
+    fn peer_rest_client_network_classifier_ignores_application_text_containing_unavailable() {
+        // Regression: a bare "unavailable" needle used to match application
+        // strings like these and take a healthy peer offline.
+        assert!(!PeerRestClient::is_network_like_error(&Error::other(
+            "peer replication statistics provider is unavailable"
+        )));
+        assert!(!PeerRestClient::is_network_like_error(&Error::other(
+            "bucket \"unavailable-logs\" not found"
+        )));
+        // Anchored renderings of a flattened Unavailable status still match:
+        // tonic >= 0.14 form ...
+        assert!(PeerRestClient::is_network_like_error(&Error::other(
+            "peer tier mutation commit RPC failed: code: 'The service is currently unavailable', message: \"peer gone\""
+        )));
+        // ... which is only anchored as long as tonic renders Unavailable this
+        // way. A tonic bump that reworded it leaves the typed path correct but
+        // this needle stale, so pin the coupling rather than discover it in a
+        // partition.
+        assert!(
+            tonic::Status::unavailable("peer gone")
+                .to_string()
+                .to_ascii_lowercase()
+                .contains("code: 'the service is currently unavailable'"),
+            "tonic reworded Code::Unavailable; update the anchored needle"
+        );
+        // ... and the tonic <= 0.13 form peers may relay in error_info.
+        assert!(PeerRestClient::is_network_like_error(&Error::other(
+            "peer tier mutation commit RPC failed: status: Unavailable, message: \"peer gone\""
+        )));
+    }
+
     #[test]
     fn tier_config_reload_outcome_keeps_tonic_and_remote_errors_typed() {
         assert!(matches!(
@@ -2193,6 +2295,14 @@ mod tests {
             tier_config_reload_connection_outcome(Error::other("can not get client, err: connection unavailable")),
             TierConfigReloadOutcome::TransientReconnect(_)
         ));
+        // The bare word is trusted only to the right of the dial-failure
+        // prefix, not anywhere in the message.
+        assert!(matches!(
+            tier_config_reload_connection_outcome(Error::other(
+                "bucket unavailable-logs rejected it, then: can not get client, err: some other reason"
+            )),
+            TierConfigReloadOutcome::Terminal(_)
+        ));
     }
 
     #[tokio::test]
@@ -2495,6 +2605,39 @@ mod tests {
         assert!(!client.offline.load(Ordering::Acquire));
     }
 
+    #[tokio::test]
+    async fn peer_rest_client_finalize_result_keeps_online_for_app_errors_mentioning_unavailable() {
+        // Regression: application error text containing "unavailable" (a
+        // remote error_info payload, or a bucket named "unavailable-logs" in
+        // a typed application status) must not take a healthy peer offline.
+        let client = test_peer_client();
+        let err = client
+            .finalize_result::<()>(Err(Error::other("peer replication statistics provider is unavailable")))
+            .await
+            .expect_err("application error should still be returned");
+        assert!(err.to_string().contains("provider is unavailable"));
+        assert!(!client.offline.load(Ordering::Acquire));
+
+        let err = client
+            .finalize_result::<()>(Err(Error::from(tonic::Status::internal(
+                "failed to reload metadata for bucket \"unavailable-logs\"",
+            ))))
+            .await
+            .expect_err("application status should still be returned");
+        assert!(err.to_string().contains("unavailable-logs"));
+        assert!(!client.offline.load(Ordering::Acquire));
+    }
+
+    #[tokio::test]
+    async fn peer_rest_client_finalize_result_marks_offline_for_typed_unavailable_status() {
+        let client = test_peer_client();
+        client
+            .finalize_result::<()>(Err(Error::from(tonic::Status::unavailable("storage layer is not initialized"))))
+            .await
+            .expect_err("network error should still be returned");
+        assert!(client.offline.load(Ordering::Acquire));
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn peer_rest_recovery_probe_logs_keep_request_id_span_context() {
         let logs = CapturedLogs::default();