fix(data-usage): make empty checks exhaustive

* ci: preserve timeout evidence for workspace tests

* ci: allow cold doctest compilation to finish

* ci: allow cold nextest compilation to finish

* ci: allow cold nextest compilation to finish

---------

Co-authored-by: houseme <housemecn@gmail.com>

.github/workflows/ci.yml

@@ -156,16 +156,69 @@ jobs:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           cache-save-if: ${{ github.ref == 'refs/heads/main' }}
 
+      - name: Prepare test evidence
+        run: |
+          mkdir -p artifacts/test-and-lint
+          {
+            echo "run_id=${GITHUB_RUN_ID}"
+            echo "job=${GITHUB_JOB}"
+            echo "runner=${RUNNER_NAME}"
+            echo "started_at=$(date --utc --iso-8601=seconds)"
+          } > artifacts/test-and-lint/run-metadata.txt
+
       # Clippy runs before the test pass: lint failures are the most common
       # CI-only breakage and should surface in minutes, not after 20+ minutes
       # of tests.
       - name: Run clippy lints
         run: cargo clippy --all-targets -- -D warnings
 
-      - name: Run tests
+      - name: Run nextest tests
         run: |
-          cargo nextest run --profile ci --all --exclude e2e_test
-          cargo test --all --doc
+          mkdir -p artifacts/test-and-lint
+          set +e
+          NEXTEST_HIDE_PROGRESS_BAR=1 timeout --verbose --signal=TERM --kill-after=30s 75m \
+            cargo nextest run --profile ci --all --exclude e2e_test \
+            --status-level all --final-status-level all \
+            2>&1 | tee artifacts/test-and-lint/nextest.log
+          status=${PIPESTATUS[0]}
+          {
+            echo "command=cargo nextest run --profile ci --all --exclude e2e_test"
+            echo "exit_status=${status}"
+            echo "finished_at=$(date --utc --iso-8601=seconds)"
+            echo
+            echo "Remaining test-related processes:"
+            pgrep -af 'cargo|nextest|target/.*/deps/' || true
+          } > artifacts/test-and-lint/nextest-diagnostics.txt
+          exit "${status}"
+
+      - name: Run documentation tests
+        run: |
+          mkdir -p artifacts/test-and-lint
+          set +e
+          timeout --verbose --signal=TERM --kill-after=30s 15m \
+            cargo test --all --doc \
+            2>&1 | tee artifacts/test-and-lint/doctest.log
+          status=${PIPESTATUS[0]}
+          {
+            echo "command=cargo test --all --doc"
+            echo "exit_status=${status}"
+            echo "finished_at=$(date --utc --iso-8601=seconds)"
+            echo
+            echo "Remaining test-related processes:"
+            pgrep -af 'cargo|rustdoc|target/.*/deps/' || true
+          } > artifacts/test-and-lint/doctest-diagnostics.txt
+          exit "${status}"
+
+      - name: Upload test reports and diagnostics
+        if: always()
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+        with:
+          name: junit-test-and-lint-${{ github.run_number }}
+          path: |
+            target/nextest/ci/junit.xml
+            artifacts/test-and-lint
+          retention-days: 3
+          if-no-files-found: error
 
       # rustfs/backlog#1289: fail if a seed rule's log anchor no longer exists
       # verbatim in the source tree (log message drifted without updating the
@@ -174,15 +227,6 @@ jobs:
       - name: Check log-analyzer rule anchors
         run: ./scripts/check_log_analyzer_rules.sh
 
-      - name: Upload test junit report
-        if: always()
-        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
-        with:
-          name: junit-test-and-lint-${{ github.run_number }}
-          path: target/nextest/ci/junit.xml
-          retention-days: 3
-          if-no-files-found: ignore
-
       # Explicit gate for migration-critical suites. These tests already ran in
       # the full nextest pass above; a single filtered nextest invocation keeps
       # the named gate without rebuilding or re-running them one package at a time.

.github/workflows/docker.yml

@@ -66,7 +66,7 @@ env:
   CARGO_TERM_COLOR: always
   REGISTRY_DOCKERHUB: rustfs/rustfs
   REGISTRY_GHCR: ghcr.io/${{ github.repository }}
-  REGISTRY_QUAY: quay.io/${{ secrets.QUAY_USERNAME }}/rustfs
+  REGISTRY_QUAY: quay.io/rustfs/rustfs
   DOCKER_PLATFORMS: linux/amd64,linux/arm64
 
 jobs:

crates/checksums/Cargo.toml

@@ -25,6 +25,9 @@ keywords = ["checksum-calculation", "verification", "integrity", "authenticity",
 categories = ["web-programming", "development-tools", "network-programming"]
 documentation = "https://docs.rs/rustfs-checksums/latest/rustfs_checksum/"
 
+[lints]
+workspace = true
+
 [dependencies]
 bytes = { workspace = true, features = ["serde"] }
 crc-fast = { workspace = true }

crates/data-usage/src/data_usage.rs

@@ -506,8 +506,35 @@ pub struct ReplicationStats {
 }
 
 impl ReplicationStats {
+    pub fn is_empty(&self) -> bool {
+        let Self {
+            pending_size,
+            replicated_size,
+            failed_size,
+            failed_count,
+            pending_count,
+            missed_threshold_size,
+            after_threshold_size,
+            missed_threshold_count,
+            after_threshold_count,
+            replicated_count,
+        } = self;
+
+        *pending_size == 0
+            && *replicated_size == 0
+            && *failed_size == 0
+            && *failed_count == 0
+            && *pending_count == 0
+            && *missed_threshold_size == 0
+            && *after_threshold_size == 0
+            && *missed_threshold_count == 0
+            && *after_threshold_count == 0
+            && *replicated_count == 0
+    }
+
+    #[deprecated(note = "use is_empty instead")]
     pub fn empty(&self) -> bool {
-        self.replicated_size == 0 && self.failed_size == 0 && self.failed_count == 0
+        self.is_empty()
     }
 }
 
@@ -520,16 +547,19 @@ pub struct ReplicationAllStats {
 }
 
 impl ReplicationAllStats {
+    pub fn is_empty(&self) -> bool {
+        let Self {
+            replica_size,
+            replica_count,
+            targets,
+        } = self;
+
+        *replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
+    }
+
+    #[deprecated(note = "use is_empty instead")]
     pub fn empty(&self) -> bool {
-        if self.replica_size != 0 && self.replica_count != 0 {
-            return false;
-        }
-        for v in self.targets.values() {
-            if !v.empty() {
-                return false;
-            }
-        }
-        true
+        self.is_empty()
     }
 }
 
@@ -783,7 +813,7 @@ impl DataUsageCache {
                     return Some(root);
                 }
                 let mut flat = self.flatten(&root);
-                if flat.replication_stats.as_ref().is_some_and(|stats| stats.empty()) {
+                if flat.replication_stats.as_ref().is_some_and(ReplicationAllStats::is_empty) {
                     flat.replication_stats = None;
                 }
                 Some(flat)
@@ -1582,6 +1612,126 @@ mod tests {
         assert_eq!(map["BETWEEN_1024B_AND_1_MB"], u64::MAX);
     }
 
+    #[test]
+    fn replication_stats_empty_checks_every_field() {
+        type SetField = fn(&mut ReplicationStats);
+
+        let cases: [(&str, SetField); 10] = [
+            ("pending_size", |stats| stats.pending_size = 1),
+            ("replicated_size", |stats| stats.replicated_size = 1),
+            ("failed_size", |stats| stats.failed_size = 1),
+            ("failed_count", |stats| stats.failed_count = 1),
+            ("pending_count", |stats| stats.pending_count = 1),
+            ("missed_threshold_size", |stats| stats.missed_threshold_size = 1),
+            ("after_threshold_size", |stats| stats.after_threshold_size = 1),
+            ("missed_threshold_count", |stats| stats.missed_threshold_count = 1),
+            ("after_threshold_count", |stats| stats.after_threshold_count = 1),
+            ("replicated_count", |stats| stats.replicated_count = 1),
+        ];
+
+        assert!(ReplicationStats::default().is_empty());
+        for (field, set_nonzero) in cases {
+            let mut stats = ReplicationStats::default();
+            set_nonzero(&mut stats);
+            assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
+        }
+    }
+
+    #[test]
+    fn replication_all_stats_empty_checks_aggregate_fields_independently() {
+        let cases = [
+            (
+                "replica_size",
+                ReplicationAllStats {
+                    replica_size: 1,
+                    ..Default::default()
+                },
+            ),
+            (
+                "replica_count",
+                ReplicationAllStats {
+                    replica_count: 1,
+                    ..Default::default()
+                },
+            ),
+        ];
+
+        assert!(ReplicationAllStats::default().is_empty());
+        for (field, stats) in cases {
+            assert!(!stats.is_empty(), "{field} must make aggregate replication stats non-empty");
+        }
+
+        let empty_targets = ReplicationAllStats {
+            targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
+            ..Default::default()
+        };
+        assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");
+
+        let stats = ReplicationAllStats {
+            targets: HashMap::from([
+                ("arn:test:empty".to_string(), ReplicationStats::default()),
+                (
+                    "arn:test:non-empty".to_string(),
+                    ReplicationStats {
+                        pending_count: 1,
+                        ..Default::default()
+                    },
+                ),
+            ]),
+            ..Default::default()
+        };
+        assert!(!stats.is_empty(), "a non-empty target must make aggregate replication stats non-empty");
+    }
+
+    #[test]
+    fn size_recursive_prunes_empty_and_preserves_pending_replication_stats() {
+        let root = hash_path("bucket");
+        let child = hash_path("bucket/child");
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(&root, &None, &DataUsageEntry::default());
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats::default()),
+                ..Default::default()
+            },
+        );
+
+        assert!(
+            cache
+                .size_recursive("bucket")
+                .expect("bucket usage should flatten")
+                .replication_stats
+                .is_none()
+        );
+
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats {
+                    targets: HashMap::from([(
+                        "arn:test:pending".to_string(),
+                        ReplicationStats {
+                            pending_count: 1,
+                            ..Default::default()
+                        },
+                    )]),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+        );
+
+        let flattened = cache.size_recursive("bucket").expect("bucket usage should flatten");
+        let replication = flattened
+            .replication_stats
+            .expect("pending-only replication stats must survive pruning");
+
+        assert_eq!(replication.targets["arn:test:pending"].pending_count, 1);
+    }
+
     #[test]
     fn test_data_usage_cache_merge_adds_missing_child() {
         let mut base = DataUsageCache::default();

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

@@ -14,7 +14,7 @@
 
 use crate::cluster::rpc::http_auth::RPC_CONTENT_SHA256_HEADER;
 use crate::cluster::rpc::{gen_tonic_signature_headers, normalize_tonic_rpc_audience};
-use crate::disk::error::{DiskError, Error as DiskErrorType};
+use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError};
 use crate::runtime::sources as runtime_sources;
 use http::Uri;
 use rustfs_protos::{
@@ -107,10 +107,79 @@ pub async fn node_service_time_out_client_no_auth(
     node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await
 }
 
+/// The typed `tonic::Status` an internode RPC failure was converted from, if
+/// this error carries one.
+pub(crate) fn embedded_tonic_status(io_err: &std::io::Error) -> Option<&tonic::Status> {
+    io_err.get_ref()?.downcast_ref::<RpcStatusError>().map(RpcStatusError::status)
+}
+
+/// Decide whether a gRPC status reports a peer we cannot currently reach,
+/// rather than an application outcome from a live peer.
+///
+/// `Unavailable` is the one code that means "no service behind this channel":
+/// the client transport raises it when the connection is broken, and the
+/// server's own not-ready gates use it deliberately.
+///
+/// `Unknown` is the client transport's escape hatch for a cause it could not
+/// map to a code — tower's "Service was not ready: <cause>", an h2 error with
+/// no gRPC mapping. Our handlers never return it, so there its message is the
+/// only evidence available and the anchored needles decide.
+///
+/// Every other code is an answer from a live peer and is never a transport
+/// failure, whatever its message says. That distinction is the point of
+/// classifying by code: a peer relaying its own downstream trouble as
+/// `Internal("connection refused ...")`, or a handler interpolating a local
+/// `io::Error` into `Status::internal`, answered us perfectly well. Marking it
+/// offline over that text is the bug this classification replaces. Likewise a
+/// `Cancelled` "Timeout expired" from the per-RPC channel deadline means the
+/// peer is slow, not gone; gating it would turn load into a partition.
+pub(crate) fn is_network_like_status(status: &tonic::Status) -> bool {
+    match status.code() {
+        tonic::Code::Unavailable => true,
+        tonic::Code::Unknown => message_has_network_needle(&status.to_string()),
+        _ => false,
+    }
+}
+
+/// Substring fallback for failures that only exist as text: dial errors
+/// wrapped by `get_client`, remote `error_info` payloads, and statuses
+/// flattened through `format!`. Needles must stay anchored to transport
+/// context — a bare word like "unavailable" also matches application text
+/// (e.g. a bucket named "unavailable-logs") and would take a healthy peer
+/// offline.
+pub(crate) fn message_has_network_needle(message: &str) -> bool {
+    let message = message.to_ascii_lowercase();
+    [
+        "temporarily offline",
+        "transport error",
+        // tonic >= 0.14 renders Code::Unavailable as
+        // `code: 'The service is currently unavailable'`.
+        "code: 'the service is currently unavailable'",
+        // RUSTFS_COMPAT_TODO(tonic-013-status-render): releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered the same status as `status: Unavailable`, and peers relay that text in error_info. Remove after the minimum supported RustFS peer version ships tonic >= 0.14.
+        "status: 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))
+}
+
 pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
     match err {
         DiskError::Timeout => true,
         DiskError::Io(io_err) => {
+            if let Some(status) = embedded_tonic_status(io_err) {
+                return is_network_like_status(status);
+            }
             if matches!(
                 io_err.kind(),
                 ErrorKind::TimedOut
@@ -124,24 +193,7 @@ pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
                 return true;
             }
 
-            let message = io_err.to_string().to_ascii_lowercase();
-            [
-                "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))
+            message_has_network_needle(&io_err.to_string())
         }
         _ => false,
     }
@@ -269,6 +321,70 @@ mod tests {
         let _ = provider.shutdown();
     }
 
+    #[test]
+    fn network_like_disk_error_uses_typed_status_code() {
+        // Transport-level Unavailable statuses justify retry/eviction.
+        assert!(is_network_like_disk_error(&DiskError::from(tonic::Status::unavailable(
+            "storage layer is not initialized"
+        ))));
+        // Application statuses from a live peer must not look network-like,
+        // even when their message contains transport-sounding words.
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::internal(
+            "failed to heal bucket \"unavailable-logs\""
+        ))));
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::unauthenticated(
+            "No valid auth token"
+        ))));
+        // A slow peer that blew the per-RPC deadline is still answering.
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::cancelled("Timeout expired"))));
+    }
+
+    #[test]
+    fn embedded_tonic_status_is_recovered_across_error_conversions() {
+        // DiskError and StorageError share one wrapper, so a status keeps its
+        // typed classification whichever error it was converted into first.
+        let from_storage: DiskErrorType = crate::error::Error::from(tonic::Status::unavailable("peer gone")).into();
+        let DiskError::Io(io_err) = &from_storage else {
+            panic!("status-derived disk error should stay an Io error");
+        };
+        assert_eq!(embedded_tonic_status(io_err).map(|status| status.code()), Some(tonic::Code::Unavailable));
+
+        let from_disk = crate::error::Error::from(DiskError::from(tonic::Status::unavailable("peer gone")));
+        let crate::error::Error::Io(io_err) = &from_disk else {
+            panic!("status-derived storage error should stay an Io error");
+        };
+        assert_eq!(embedded_tonic_status(io_err).map(|status| status.code()), Some(tonic::Code::Unavailable));
+    }
+
+    #[test]
+    fn network_like_disk_error_ignores_transport_words_in_application_statuses() {
+        // Same contract as the peer client: a status the peer answered with
+        // is not a transport failure, so it must not drive a reconnect even
+        // when its message describes one.
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::internal(
+            "connection refused while dialing downstream backend"
+        ))));
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::unauthenticated(
+            "connection reset while validating token"
+        ))));
+    }
+
+    #[test]
+    fn network_like_disk_error_requires_anchored_unavailable_needle() {
+        // Regression: a bare "unavailable" needle used to match application
+        // text such as a bucket name.
+        assert!(!is_network_like_disk_error(&DiskError::other("bucket \"unavailable-logs\" not found")));
+        // Anchored renderings of a flattened Unavailable status still match.
+        assert!(is_network_like_disk_error(&DiskError::other(
+            "code: 'The service is currently unavailable', message: \"peer gone\""
+        )));
+        assert!(is_network_like_disk_error(&DiskError::other(
+            "status: Unavailable, message: \"peer gone\""
+        )));
+        assert!(is_network_like_disk_error(&DiskError::other("connection refused")));
+        assert!(!is_network_like_disk_error(&DiskError::FileNotFound));
+    }
+
     #[test]
     fn test_signature_interceptor_keeps_auth_headers() {
         ensure_test_rpc_secret();

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();

crates/ecstore/src/disk/error.rs

@@ -337,9 +337,54 @@ impl From<DiskError> for std::io::Error {
     }
 }
 
+/// The single in-band representation of a failed internode RPC: it carries the
+/// typed `tonic::Status` so failure classifiers can read the gRPC code instead
+/// of substring-matching the rendered message (see `is_network_like_status`).
+/// Both `DiskError` and `StorageError` wrap statuses in this type, so one
+/// downcast recovers the status regardless of which error the status was
+/// converted into first.
+pub(crate) struct RpcStatusError(tonic::Status);
+
+impl RpcStatusError {
+    pub(crate) fn status(&self) -> &tonic::Status {
+        &self.0
+    }
+}
+
+impl From<tonic::Status> for RpcStatusError {
+    fn from(status: tonic::Status) -> Self {
+        Self(status)
+    }
+}
+
+impl std::fmt::Display for RpcStatusError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Display::fmt(&self.0, f)
+    }
+}
+
+/// `tonic::Status`'s own `Debug` prints its `MetadataMap`, i.e. every response
+/// header and trailer the peer sent. Those are remote-controlled and can carry
+/// credentials injected by a proxy in front of the peer, so keep them out of
+/// anything that reaches a log.
+impl std::fmt::Debug for RpcStatusError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RpcStatusError")
+            .field("code", &self.0.code())
+            .field("message", &self.0.message())
+            .finish()
+    }
+}
+
+impl StdError for RpcStatusError {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        Some(&self.0)
+    }
+}
+
 impl From<tonic::Status> for DiskError {
     fn from(e: tonic::Status) -> Self {
-        DiskError::other(e.message().to_string())
+        DiskError::Io(io::Error::other(RpcStatusError(e)))
     }
 }
 

crates/ecstore/src/error/mod.rs

@@ -818,7 +818,11 @@ impl From<s3s::xml::DeError> for Error {
 
 impl From<tonic::Status> for Error {
     fn from(e: tonic::Status) -> Self {
-        Error::other(e.to_string())
+        // Keep the typed status as the io::Error payload instead of a
+        // flattened string so RPC failure classifiers can read the gRPC code
+        // via downcast (see `PeerRestClient::is_network_like_error`).
+        // `RpcStatusError` renders exactly as `e.to_string()` did.
+        Error::other(crate::disk::error::RpcStatusError::from(e))
     }
 }
 

crates/scanner/src/data_usage_define.rs

@@ -698,7 +698,7 @@ impl DataUsageCache {
                 let mut visited = HashSet::new();
                 visited.insert(hash_path(path).key());
                 let mut flat = self.flatten_with_guard(root, &mut visited, 0);
-                if flat.replication_stats.as_ref().is_some_and(|stats| stats.empty()) {
+                if flat.replication_stats.as_ref().is_some_and(|stats| stats.is_empty()) {
                     flat.replication_stats = None;
                 }
                 Some(flat)
@@ -1574,6 +1574,7 @@ mod tests {
     use super::*;
     use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
     use crate::{ScannerGetObjectReader, ScannerPutObjReader};
+    use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
     use serde_json::Value;
     use std::io::Cursor;
     use std::pin::Pin;
@@ -2767,6 +2768,55 @@ mod tests {
         assert!(flat.children.is_empty());
     }
 
+    #[test]
+    fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
+        let root = hash_path("bucket");
+        let child = hash_path("bucket/child");
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(&root, &None, &DataUsageEntry::default());
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats::default()),
+                ..Default::default()
+            },
+        );
+
+        assert!(
+            cache
+                .size_recursive("bucket")
+                .expect("scanner bucket usage should flatten")
+                .replication_stats
+                .is_none()
+        );
+
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats {
+                    targets: HashMap::from([(
+                        "arn:test:threshold".to_string(),
+                        ReplicationStats {
+                            after_threshold_count: 1,
+                            ..Default::default()
+                        },
+                    )]),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+        );
+
+        let flattened = cache.size_recursive("bucket").expect("scanner bucket usage should flatten");
+        let replication = flattened
+            .replication_stats
+            .expect("threshold-only replication stats must survive pruning");
+
+        assert_eq!(replication.targets["arn:test:threshold"].after_threshold_count, 1);
+    }
+
     #[test]
     fn checked_flatten_rejects_dangling_child() {
         let root_key = hash_path("bucket").key();

docs/architecture/compat-cleanup-register.md

@@ -19,6 +19,7 @@ for later deletion.
 - `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.
 - `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus.
 - `backlog-1316` legacy encrypted multipart range seek: the feature remains opt-in until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently.
+- `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
 - `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.
 - `sse-local-dek-json-v1` legacy local SSE DEK decoding: releases before the JSON envelope wrote wrapped DEKs as `base64(nonce):base64(ciphertext)`, so readers retain that decoder while all new writes use the versioned JSON envelope. Remove the colon decoder after the minimum supported direct-upgrade release writes JSON envelopes and migration tooling has rewritten every retained legacy object.