Merge branch 'main' into houseme/backlog-1479-durable-transition-jobs

crates/config/src/constants/internode.rs

@@ -118,6 +118,24 @@ pub const DEFAULT_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED: bool = false;
 const _: () = assert!(!DEFAULT_INTERNODE_RPC_MSGPACK_ONLY);
 const _: () = assert!(!DEFAULT_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED);
 
+/// Require target-bound v2 signatures on every internode gRPC request, rejecting the legacy
+/// constant-target fallback instead of accepting it (<https://github.com/rustfs/backlog/issues/1327>).
+///
+/// Defaults to `false` (fail-open): a request without any v2 auth headers keeps authenticating
+/// through the legacy signature, so legacy-only peers survive rolling upgrades with byte-for-byte
+/// the pre-gate acceptance behavior. This is a rollout lever, not a wire-format change: it may only
+/// be enabled **after** the v1-fallback counter
+/// (`rustfs_system_network_internode_signature_v1_fallback_total`) has read zero across a release
+/// window fleet-wide, confirming every peer already sends v2 authentication on every internode gRPC
+/// request. Single-env rollback. Requests that do carry v2 headers are unaffected by this switch:
+/// they are always verified as v2 with no downgrade, strict or not.
+pub const ENV_INTERNODE_RPC_SIGNATURE_STRICT: &str = "RUSTFS_INTERNODE_RPC_SIGNATURE_STRICT";
+pub const DEFAULT_INTERNODE_RPC_SIGNATURE_STRICT: bool = false;
+
+// Compile-time invariant: fail-open by default so legacy-only peers keep authenticating during
+// rolling upgrades until the fleet-wide v1-fallback counter reads zero.
+const _: () = assert!(!DEFAULT_INTERNODE_RPC_SIGNATURE_STRICT);
+
 /// Consecutive-failure threshold after which an internode peer is marked offline (grpc-optimization
 /// P3 observability).
 ///
@@ -288,6 +306,12 @@ mod tests {
         );
     }
 
+    #[test]
+    fn internode_signature_strict_env_name_is_stable() {
+        // The fail-open default invariant is asserted at compile time next to the definition.
+        assert_eq!(ENV_INTERNODE_RPC_SIGNATURE_STRICT, "RUSTFS_INTERNODE_RPC_SIGNATURE_STRICT");
+    }
+
     #[test]
     fn internode_offline_failure_threshold_defaults_and_env_name() {
         assert_eq!(DEFAULT_INTERNODE_OFFLINE_FAILURE_THRESHOLD, 3);

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

@@ -35,6 +35,8 @@ use http::{HeaderMap, HeaderValue, Method, Uri};
 #[cfg(test)]
 use rustfs_credentials::{DEFAULT_SECRET_KEY, RPC_SECRET_REQUIRED_MESSAGE};
 use rustfs_credentials::{RPC_SECRET_REQUIRED_OPERATOR_MESSAGE, try_get_rpc_token};
+use rustfs_io_metrics::internode_metrics::global_internode_metrics;
+use rustfs_utils::get_env_bool;
 use sha2::Digest as _;
 use sha2::Sha256;
 use std::collections::{HashSet, VecDeque};
@@ -60,6 +62,12 @@ const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes
 const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(601);
 const MAX_REPLAY_PROTECTED_NONCES: usize = 65_536;
 pub const TONIC_RPC_PREFIX: &str = "/node_service.NodeService";
+static INTERNODE_RPC_SIGNATURE_STRICT: LazyLock<bool> = LazyLock::new(|| {
+    get_env_bool(
+        rustfs_config::ENV_INTERNODE_RPC_SIGNATURE_STRICT,
+        rustfs_config::DEFAULT_INTERNODE_RPC_SIGNATURE_STRICT,
+    )
+});
 static RPC_SECRET_RESOLUTION_LOG_ONCE: Once = Once::new();
 
 #[derive(Default)]
@@ -412,12 +420,40 @@ fn has_v2_auth_headers(headers: &HeaderMap) -> bool {
     .any(|name| headers.contains_key(*name))
 }
 
+/// Whether the server requires target-bound v2 authentication on every internode gRPC request,
+/// rejecting the legacy constant-target fallback instead of accepting it. Default-off rollout
+/// lever gated on the v1-fallback counter reading zero fleet-wide; see
+/// [`rustfs_config::ENV_INTERNODE_RPC_SIGNATURE_STRICT`] and
+/// <https://github.com/rustfs/backlog/issues/1327>.
+fn internode_rpc_signature_strict() -> bool {
+    *INTERNODE_RPC_SIGNATURE_STRICT
+}
+
 /// Verify gRPC authentication, preferring v2 without downgrade on malformed v2 metadata.
 pub fn verify_tonic_rpc_signature(audience: &str, path: &str, headers: &HeaderMap) -> std::io::Result<()> {
+    verify_tonic_rpc_signature_with_strictness(audience, path, headers, internode_rpc_signature_strict())
+}
+
+/// [`verify_tonic_rpc_signature`] with the strict gate injected as a parameter, so both rollout
+/// postures are unit-testable without racing on process-global environment variables.
+fn verify_tonic_rpc_signature_with_strictness(
+    audience: &str,
+    path: &str,
+    headers: &HeaderMap,
+    strict: bool,
+) -> std::io::Result<()> {
     if !has_v2_auth_headers(headers) {
         // RUSTFS_COMPAT_TODO(heal-rpc-auth-v2): accept old peers during rolling upgrades. Remove after the minimum
         // supported RustFS peer version sends v2 authentication on every internode gRPC request.
-        return verify_rpc_signature(TONIC_RPC_PREFIX, &Method::GET, headers);
+        if strict {
+            return Err(std::io::Error::other("RPC v2 authentication required"));
+        }
+        verify_rpc_signature(TONIC_RPC_PREFIX, &Method::GET, headers)?;
+        // Count only ACCEPTED legacy-only requests: this counter is the convergence gate that must
+        // read zero fleet-wide across a release window before
+        // `RUSTFS_INTERNODE_RPC_SIGNATURE_STRICT` may be enabled.
+        global_internode_metrics().record_signature_v1_fallback();
+        return Ok(());
     }
 
     let path = path
@@ -1111,7 +1147,10 @@ mod tests {
         assert_eq!(error.to_string(), "Invalid RPC v2 signature");
     }
 
+    // The `rpc_v1_fallback_counter` serial group covers every test that drives (or asserts on) the
+    // process-global v1-fallback counter, so exact-delta assertions cannot race with each other.
     #[test]
+    #[serial_test::serial(rpc_v1_fallback_counter)]
     fn legacy_tonic_signature_remains_accepted_during_rolling_upgrade() {
         ensure_test_rpc_secret();
         let headers = gen_signature_headers(TONIC_RPC_PREFIX, &Method::GET).expect("legacy auth headers should build");
@@ -1119,6 +1158,87 @@ mod tests {
         assert!(verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/Ping", &headers).is_ok());
     }
 
+    #[test]
+    #[serial_test::serial(rpc_v1_fallback_counter)]
+    fn accepted_legacy_fallback_increments_v1_fallback_counter() {
+        ensure_test_rpc_secret();
+        let headers = gen_signature_headers(TONIC_RPC_PREFIX, &Method::GET).expect("legacy auth headers should build");
+        let before = global_internode_metrics().snapshot().signature_v1_fallback_total;
+
+        assert!(
+            verify_tonic_rpc_signature_with_strictness("node-a:9000", "/node_service.NodeService/Ping", &headers, false).is_ok(),
+            "a legacy-only peer must keep authenticating while the strict gate is off"
+        );
+
+        let after = global_internode_metrics().snapshot().signature_v1_fallback_total;
+        assert_eq!(
+            after,
+            before + 1,
+            "an accepted legacy-only request must increment the v1 fallback counter exactly once"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(rpc_v1_fallback_counter)]
+    fn rejected_legacy_fallback_does_not_count_as_v1_fallback() {
+        ensure_test_rpc_secret();
+        // Legacy-shaped headers with a forged signature: the fallback path runs but must reject,
+        // and a rejected request is not a rollout-convergence signal.
+        let mut headers = HeaderMap::new();
+        let now = OffsetDateTime::now_utc().unix_timestamp();
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_static("not-a-real-signature"));
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&now.to_string()).unwrap());
+        let before = global_internode_metrics().snapshot().signature_v1_fallback_total;
+
+        assert!(
+            verify_tonic_rpc_signature_with_strictness("node-a:9000", "/node_service.NodeService/Ping", &headers, false).is_err(),
+            "a forged legacy signature must still be rejected"
+        );
+
+        let after = global_internode_metrics().snapshot().signature_v1_fallback_total;
+        assert_eq!(after, before, "a rejected legacy request must not count as an accepted fallback");
+    }
+
+    #[test]
+    #[serial_test::serial(rpc_v1_fallback_counter)]
+    fn strict_gate_rejects_legacy_only_auth_but_keeps_v2() {
+        ensure_test_rpc_secret();
+        let legacy = gen_signature_headers(TONIC_RPC_PREFIX, &Method::GET).expect("legacy auth headers should build");
+        let before = global_internode_metrics().snapshot().signature_v1_fallback_total;
+        let error = verify_tonic_rpc_signature_with_strictness("node-a:9000", "/node_service.NodeService/Ping", &legacy, true)
+            .expect_err("strict mode must reject legacy-only authentication");
+        assert_eq!(error.to_string(), "RPC v2 authentication required");
+
+        let v2 = gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "Ping", None)
+            .expect("tonic auth headers should build");
+        assert!(
+            verify_tonic_rpc_signature_with_strictness("node-a:9000", "/node_service.NodeService/Ping", &v2, true).is_ok(),
+            "strict mode must keep accepting v2-authenticated peers"
+        );
+        let after = global_internode_metrics().snapshot().signature_v1_fallback_total;
+        assert_eq!(after, before, "neither a strict rejection nor a v2 acceptance is a legacy fallback");
+    }
+
+    #[test]
+    #[serial_test::serial(rpc_v1_fallback_counter)]
+    fn strict_gate_default_posture_is_fail_open_legacy_accept() {
+        ensure_test_rpc_secret();
+        // The public entry point resolves strictness from the environment, whose compile-time
+        // default is pinned to false in `rustfs_config`. A legacy-only peer therefore keeps
+        // authenticating through the default build with no configuration at all.
+        let headers = gen_signature_headers(TONIC_RPC_PREFIX, &Method::GET).expect("legacy auth headers should build");
+        assert!(
+            verify_tonic_rpc_signature_with_strictness(
+                "node-a:9000",
+                "/node_service.NodeService/Ping",
+                &headers,
+                rustfs_config::DEFAULT_INTERNODE_RPC_SIGNATURE_STRICT,
+            )
+            .is_ok(),
+            "the default strict posture must accept legacy-only peers"
+        );
+    }
+
     #[test]
     fn body_bound_tonic_request_rejects_replay_and_body_tampering() {
         ensure_test_rpc_secret();

crates/ecstore/src/diagnostics/get.rs

@@ -14,8 +14,8 @@
 
 use crate::disk::error::DiskError;
 use crate::error::StorageError;
-use std::io;
 use std::time::Instant;
+use std::{error::Error as StdError, fmt, io};
 
 pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming";
 pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine";
@@ -207,9 +207,42 @@ pub(crate) fn classify_disk_error(err: &DiskError) -> GetObjectFailureReason {
     }
 }
 
-pub(crate) fn classify_io_error(err: &io::Error) -> GetObjectFailureReason {
+#[derive(Debug)]
+struct GetObjectDownstreamClosed {
+    source: io::Error,
+}
+
+impl fmt::Display for GetObjectDownstreamClosed {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "GetObject output closed: {}", self.source)
+    }
+}
+
+impl StdError for GetObjectDownstreamClosed {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        Some(&self.source)
+    }
+}
+
+pub(crate) fn mark_get_object_downstream_closed(err: io::Error) -> io::Error {
+    match err.kind() {
+        io::ErrorKind::BrokenPipe | io::ErrorKind::ConnectionReset | io::ErrorKind::ConnectionAborted => {
+            io::Error::new(err.kind(), GetObjectDownstreamClosed { source: err })
+        }
+        _ => err,
+    }
+}
+
+pub(crate) fn classify_io_error(err: &io::Error) -> GetObjectFailureReason {
+    let mut source = err.get_ref().map(|source| source as &(dyn StdError + 'static));
+    while let Some(error) = source {
+        if error.is::<GetObjectDownstreamClosed>() {
+            return GetObjectFailureReason::DownstreamClosed;
+        }
+        source = error.source();
+    }
+
     match err.kind() {
-        io::ErrorKind::BrokenPipe | io::ErrorKind::ConnectionReset => GetObjectFailureReason::DownstreamClosed,
         io::ErrorKind::TimedOut => GetObjectFailureReason::Timeout,
         io::ErrorKind::UnexpectedEof => GetObjectFailureReason::ShortRead,
         io::ErrorKind::InvalidInput | io::ErrorKind::InvalidData => GetObjectFailureReason::RangeOrLengthInvalid,
@@ -260,6 +293,16 @@ mod tests {
             classify_storage_error(&StorageError::InvalidRangeSpec("bad range".to_string())),
             GetObjectFailureReason::RangeOrLengthInvalid
         );
+
+        let internal_broken_pipe = StorageError::Io(io::Error::from(io::ErrorKind::BrokenPipe));
+        assert_eq!(classify_storage_error(&internal_broken_pipe), GetObjectFailureReason::Io);
+
+        let downstream_closed = StorageError::Io(mark_get_object_downstream_closed(io::Error::from(io::ErrorKind::BrokenPipe)));
+        assert_eq!(
+            classify_storage_error(&downstream_closed),
+            GetObjectFailureReason::DownstreamClosed,
+            "the legacy duplex producer must suppress only explicitly tagged output closes"
+        );
     }
 
     #[test]
@@ -277,8 +320,8 @@ mod tests {
     #[test]
     fn classifies_io_errors_for_get_pipeline() {
         let cases = [
-            (io::ErrorKind::BrokenPipe, GetObjectFailureReason::DownstreamClosed),
-            (io::ErrorKind::ConnectionReset, GetObjectFailureReason::DownstreamClosed),
+            (io::ErrorKind::BrokenPipe, GetObjectFailureReason::Io),
+            (io::ErrorKind::ConnectionReset, GetObjectFailureReason::Io),
             (io::ErrorKind::TimedOut, GetObjectFailureReason::Timeout),
             (io::ErrorKind::UnexpectedEof, GetObjectFailureReason::ShortRead),
             (io::ErrorKind::InvalidInput, GetObjectFailureReason::RangeOrLengthInvalid),
@@ -292,6 +335,18 @@ mod tests {
         }
     }
 
+    #[test]
+    fn classifies_marked_output_close_as_downstream_closed() {
+        for kind in [
+            io::ErrorKind::BrokenPipe,
+            io::ErrorKind::ConnectionReset,
+            io::ErrorKind::ConnectionAborted,
+        ] {
+            let err = mark_get_object_downstream_closed(io::Error::from(kind));
+            assert_eq!(classify_io_error(&err), GetObjectFailureReason::DownstreamClosed, "kind={kind:?}");
+        }
+    }
+
     #[test]
     fn keeps_metric_labels_stable() {
         assert_eq!(GetObjectFailureReason::ReadQuorum.as_str(), "read_quorum");

crates/ecstore/src/set_disk/mod.rs

@@ -4446,6 +4446,7 @@ mod tests {
     use crate::layout::endpoints::SetupType;
     use crate::object_api::BLOCK_SIZE_V2;
     use crate::object_api::ObjectInfo;
+    use crate::set_disk::read::GetObjectOutputKind;
     use crate::storage_api_contracts::{
         heal::HealOperations as _, lifecycle::TransitionedObject, list::ListOperations as _, multipart::CompletePart,
         namespace::NamespaceLocking as _, object::ObjectIO as _, object::ObjectOperations as _,
@@ -8400,6 +8401,7 @@ mod tests {
                 range_offset,
                 range_length as i64,
                 &mut writer,
+                GetObjectOutputKind::HttpResponse,
                 fi,
                 files,
                 &disks,
@@ -8511,6 +8513,7 @@ mod tests {
                 0,
                 total_size as i64,
                 &mut writer,
+                GetObjectOutputKind::HttpResponse,
                 fi,
                 files,
                 &disks,

crates/ecstore/src/set_disk/ops/object.rs

@@ -21,6 +21,7 @@
 
 use super::super::*;
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
+use crate::set_disk::read::GetObjectOutputKind;
 
 use crate::bucket::lifecycle::{
     tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry},
@@ -31,6 +32,7 @@ use crate::bucket::lifecycle::{
         save_transition_transaction_record,
     },
 };
+use crate::diagnostics::get::GetObjectFailureReason;
 use crate::disk::OldCurrentSize;
 use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed};
 use crate::services::tier::tier::{TierConfigMgr, TierOperationLease};
@@ -522,6 +524,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 0,
                 object_info.size,
                 &mut output,
+                GetObjectOutputKind::Internal,
                 fi,
                 files,
                 &disks,
@@ -650,6 +653,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 offset,
                 length,
                 &mut writer,
+                GetObjectOutputKind::HttpResponse,
                 fi,
                 files,
                 &disks,
@@ -664,24 +668,43 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             .await
             {
                 let reason = classify_storage_error(&e);
-                record_get_object_pipeline_failure(GET_STAGE_EMIT, reason);
-                error!(
-                    event = EVENT_SET_DISK_WRITE,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_SET_DISK,
-                    bucket,
-                    object,
-                    pool_index,
-                    set_index,
-                    offset,
-                    requested_length = length,
-                    skip_verify_bitrot = skip_verify,
-                    state = "read_pipeline_failed",
-                    stage = GET_STAGE_EMIT,
-                    reason = reason.as_str(),
-                    error = ?e,
-                    "Set disk object read pipeline failed"
-                );
+                if reason == GetObjectFailureReason::DownstreamClosed {
+                    debug!(
+                        event = EVENT_SET_DISK_WRITE,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        bucket,
+                        object,
+                        pool_index,
+                        set_index,
+                        offset,
+                        requested_length = length,
+                        state = "downstream_closed",
+                        stage = GET_STAGE_EMIT,
+                        reason = reason.as_str(),
+                        error = ?e,
+                        "Set disk object read pipeline stopped after downstream closed"
+                    );
+                } else {
+                    record_get_object_pipeline_failure(GET_STAGE_EMIT, reason);
+                    error!(
+                        event = EVENT_SET_DISK_WRITE,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        bucket,
+                        object,
+                        pool_index,
+                        set_index,
+                        offset,
+                        requested_length = length,
+                        skip_verify_bitrot = skip_verify,
+                        state = "read_pipeline_failed",
+                        stage = GET_STAGE_EMIT,
+                        reason = reason.as_str(),
+                        error = ?e,
+                        "Set disk object read pipeline failed"
+                    );
+                }
             };
         });
 
@@ -3356,6 +3379,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 0,
                 cloned_fi.size,
                 &mut writer,
+                GetObjectOutputKind::Internal,
                 cloned_fi,
                 meta_arr,
                 &online_disks,

crates/ecstore/src/set_disk/read.rs

@@ -34,8 +34,8 @@ use crate::diagnostics::get::{
     GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GET_STAGE_READER_SETUP_DROP_PENDING,
     GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM, GET_STAGE_READER_TASK_BITROT_READER_INIT,
     GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION, GetObjectFailureReason, classify_disk_error,
-    get_stage_timer_if_enabled, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path,
-    record_get_stage_duration_if_enabled,
+    get_stage_timer_if_enabled, mark_get_object_downstream_closed, record_get_object_pipeline_failure,
+    record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
 };
 use crate::erasure::coding::BitrotReader;
 use crate::io_support::bitrot::{
@@ -53,12 +53,60 @@ use std::{
     task::{Context, Poll},
     time::{Duration, Instant},
 };
-use tokio::io::{AsyncRead, ReadBuf};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::RwLock;
 use tokio::task::JoinSet;
 
 use super::core::io_primitives::*;
 
+#[derive(Clone, Copy)]
+pub(super) enum GetObjectOutputKind {
+    HttpResponse,
+    Internal,
+}
+
+struct GetObjectOutputWriter<'a, W> {
+    inner: &'a mut W,
+    output_kind: GetObjectOutputKind,
+}
+
+impl<'a, W> GetObjectOutputWriter<'a, W> {
+    fn new(inner: &'a mut W, output_kind: GetObjectOutputKind) -> Self {
+        Self { inner, output_kind }
+    }
+}
+
+fn map_get_object_output_error(output_kind: GetObjectOutputKind, err: std::io::Error) -> std::io::Error {
+    if matches!(output_kind, GetObjectOutputKind::HttpResponse) {
+        mark_get_object_downstream_closed(err)
+    } else {
+        err
+    }
+}
+
+impl<W: AsyncWrite + Unpin> AsyncWrite for GetObjectOutputWriter<'_, W> {
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
+        let output_kind = self.output_kind;
+        Pin::new(&mut *self.inner)
+            .poll_write(cx, buf)
+            .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err)))
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let output_kind = self.output_kind;
+        Pin::new(&mut *self.inner)
+            .poll_flush(cx)
+            .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err)))
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let output_kind = self.output_kind;
+        Pin::new(&mut *self.inner)
+            .poll_shutdown(cx)
+            .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err)))
+    }
+}
+
 impl SetDisks {
     async fn get_object_metadata_cache_bypass_reason(
         &self,
@@ -586,6 +634,7 @@ impl SetDisks {
         offset: usize,
         length: i64,
         writer: &mut W,
+        output_kind: GetObjectOutputKind,
         fi: FileInfo,
         files: Vec<FileInfo>,
         disks: &[Option<DiskStore>],
@@ -969,9 +1018,10 @@ impl SetDisks {
             let unattempted_data_shards = !reader_setup.data_shards_attempted(erasure.data_shards);
             let readers = reader_setup.readers;
             let deferred_stripe_handles = reader_setup.deferred_stripe_handles;
+            let mut output_writer = GetObjectOutputWriter::new(writer, output_kind);
             let (written, err) = erasure
                 .decode_with_stripe_handles(
-                    writer,
+                    &mut output_writer,
                     readers,
                     part_offset,
                     part_length,
@@ -989,7 +1039,7 @@ impl SetDisks {
                 metrics_size_bucket,
                 decode_elapsed.as_secs_f64(),
             );
-            if decode_elapsed >= SLOW_OBJECT_READ_LOG_THRESHOLD || err.is_some() {
+            if decode_elapsed >= SLOW_OBJECT_READ_LOG_THRESHOLD && err.is_none() {
                 warn!(
                     event = EVENT_SET_DISK_READ,
                     component = LOG_COMPONENT_ECSTORE,
@@ -1005,9 +1055,8 @@ impl SetDisks {
                     available_shards,
                     missing_shards,
                     elapsed_ms = decode_elapsed.as_millis(),
-                    error = ?err,
-                    state = if err.is_some() { "decode_failed_or_slow" } else { "decode_slow" },
-                    "Set disk object decode stage is slow or failed"
+                    state = "decode_slow",
+                    "Set disk object decode stage is slow"
                 );
             }
             debug!(
@@ -1051,7 +1100,7 @@ impl SetDisks {
                 if has_err {
                     let reason = classify_disk_error(&de_err);
                     if reason == GetObjectFailureReason::DownstreamClosed {
-                        warn!(
+                        debug!(
                             bucket,
                             object,
                             part_index = current_part,
@@ -1969,6 +2018,7 @@ mod metadata_cache_tests {
             2,
             1,
             &mut output,
+            GetObjectOutputKind::Internal,
             valid_test_fileinfo(object),
             Vec::new(),
             &[],
@@ -1992,6 +2042,7 @@ mod metadata_cache_tests {
             usize::MAX,
             1,
             &mut output,
+            GetObjectOutputKind::Internal,
             overflow,
             Vec::new(),
             &[],
@@ -2013,6 +2064,7 @@ mod metadata_cache_tests {
             1,
             1,
             &mut output,
+            GetObjectOutputKind::Internal,
             valid_test_fileinfo(object),
             Vec::new(),
             &[],
@@ -2036,6 +2088,7 @@ mod metadata_cache_tests {
             0,
             1,
             &mut output,
+            GetObjectOutputKind::Internal,
             invalid_erasure,
             Vec::new(),
             &[],
@@ -2079,6 +2132,7 @@ mod metadata_cache_tests {
             0,
             1,
             &mut output,
+            GetObjectOutputKind::Internal,
             fi,
             Vec::new(),
             &[],
@@ -2920,11 +2974,60 @@ mod tests {
         Arc,
         atomic::{AtomicUsize, Ordering},
     };
-    use tokio::io::AsyncReadExt;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
     const CODEC_STREAMING_TEST_BUCKET: &str = "bucket";
     const CODEC_STREAMING_TEST_OBJECT: &str = "object";
 
+    struct ClosedOutputWriter;
+
+    impl AsyncWrite for ClosedOutputWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<std::io::Result<usize>> {
+            Poll::Ready(Err(std::io::Error::from(ErrorKind::BrokenPipe)))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[tokio::test]
+    async fn output_writer_marks_closed_duplex_reader_as_downstream_close() {
+        let (reader, mut inner) = tokio::io::duplex(64);
+        drop(reader);
+        let mut writer = GetObjectOutputWriter::new(&mut inner, GetObjectOutputKind::HttpResponse);
+        let err = writer
+            .write_all(b"range payload")
+            .await
+            .expect_err("closed duplex reader must fail the output write");
+
+        assert_eq!(
+            classify_disk_error(&DiskError::from(err)),
+            GetObjectFailureReason::DownstreamClosed,
+            "only the output adapter may classify a broken pipe as a downstream close"
+        );
+    }
+
+    #[tokio::test]
+    async fn output_writer_preserves_remote_broken_pipe_as_io_failure() {
+        let mut inner = ClosedOutputWriter;
+        let mut writer = GetObjectOutputWriter::new(&mut inner, GetObjectOutputKind::Internal);
+        let err = writer
+            .write_all(b"transition payload")
+            .await
+            .expect_err("closed remote writer must fail the output write");
+
+        assert_eq!(
+            classify_disk_error(&DiskError::from(err)),
+            GetObjectFailureReason::Io,
+            "remote transition writer failures must not be mistaken for HTTP client cancellation"
+        );
+    }
+
     async fn local_test_disks(count: usize, bucket: &str) -> (Vec<tempfile::TempDir>, Vec<Option<crate::disk::DiskStore>>) {
         let mut dirs = Vec::with_capacity(count);
         let mut disks = Vec::with_capacity(count);
@@ -4019,6 +4122,7 @@ mod tests {
             0,
             part_data.len() as i64,
             &mut output,
+            GetObjectOutputKind::Internal,
             fi,
             files,
             &disks,

crates/io-metrics/src/internode_metrics.rs

@@ -59,6 +59,7 @@ const INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL: &str =
 const INTERNODE_OPERATION_PAYLOAD_BYTES: &str = "rustfs_system_network_internode_operation_payload_bytes";
 const INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL: &str = "rustfs_system_network_internode_operation_large_payloads_total";
 const INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_fallback_total";
+const INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_signature_v1_fallback_total";
 const ERASURE_WRITE_QUORUM_FAILURES_TOTAL: &str = "rustfs_system_storage_erasure_write_quorum_failures_total";
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -148,6 +149,7 @@ pub struct InternodeMetricsSnapshot {
     pub operation_http_versions_total: u64,
     pub operation_stall_timeouts_total: u64,
     pub operation_write_shutdown_errors_total: u64,
+    pub signature_v1_fallback_total: u64,
 }
 
 #[derive(Debug, Default)]
@@ -164,6 +166,7 @@ pub struct InternodeMetrics {
     operation_http_versions_total: AtomicU64,
     operation_stall_timeouts_total: AtomicU64,
     operation_write_shutdown_errors_total: AtomicU64,
+    signature_v1_fallback_total: AtomicU64,
 }
 
 impl InternodeMetrics {
@@ -360,6 +363,17 @@ impl InternodeMetrics {
         counter!(INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, DIRECTION_LABEL => direction, MESSAGE_LABEL => message).increment(1);
     }
 
+    /// Count an internode gRPC request that was accepted through the legacy constant-target
+    /// signature because it carried no v2 auth headers (rolling-upgrade fallback, see
+    /// <https://github.com/rustfs/backlog/issues/1327>). Only accepted requests count: rejected
+    /// requests never authenticated, so they are not a rollout signal. This counter must read zero
+    /// across a release window fleet-wide before `RUSTFS_INTERNODE_RPC_SIGNATURE_STRICT` may be
+    /// enabled; after the strict flip the legacy fallback path is closed and the counter stays flat.
+    pub fn record_signature_v1_fallback(&self) {
+        self.signature_v1_fallback_total.fetch_add(1, Ordering::Relaxed);
+        counter!(INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL).increment(1);
+    }
+
     pub fn record_erasure_write_quorum_failure(&self, stage: &'static str, dominant_error: &'static str) {
         counter!(
             ERASURE_WRITE_QUORUM_FAILURES_TOTAL,
@@ -406,6 +420,7 @@ impl InternodeMetrics {
             operation_http_versions_total: self.operation_http_versions_total.load(Ordering::Relaxed),
             operation_stall_timeouts_total: self.operation_stall_timeouts_total.load(Ordering::Relaxed),
             operation_write_shutdown_errors_total: self.operation_write_shutdown_errors_total.load(Ordering::Relaxed),
+            signature_v1_fallback_total: self.signature_v1_fallback_total.load(Ordering::Relaxed),
         }
     }
 
@@ -423,6 +438,7 @@ impl InternodeMetrics {
         self.operation_http_versions_total.store(0, Ordering::Relaxed);
         self.operation_stall_timeouts_total.store(0, Ordering::Relaxed);
         self.operation_write_shutdown_errors_total.store(0, Ordering::Relaxed);
+        self.signature_v1_fallback_total.store(0, Ordering::Relaxed);
     }
 }
 
@@ -727,6 +743,10 @@ mod tests {
         );
         assert_eq!(INTERNODE_MSGPACK_DIRECTION_REQUEST, "request");
         assert_eq!(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "response");
+        assert_eq!(
+            INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL,
+            "rustfs_system_network_internode_signature_v1_fallback_total"
+        );
     }
 
     #[test]
@@ -737,6 +757,20 @@ mod tests {
         metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo");
     }
 
+    #[test]
+    fn signature_v1_fallback_counter_updates_snapshot_and_resets() {
+        // Instance-local metrics keep this independent of the process-global registry.
+        let metrics = InternodeMetrics::default();
+        assert_eq!(metrics.snapshot().signature_v1_fallback_total, 0);
+
+        metrics.record_signature_v1_fallback();
+        metrics.record_signature_v1_fallback();
+        assert_eq!(metrics.snapshot().signature_v1_fallback_total, 2);
+
+        metrics.reset_for_test();
+        assert_eq!(metrics.snapshot().signature_v1_fallback_total, 0);
+    }
+
     #[test]
     fn cluster_peer_flips_offline_after_threshold_and_back_online() {
         // Unique addr keeps this independent of the process-global registry / other tests.

rustfs/src/app/object_usecase.rs

@@ -1500,11 +1500,7 @@ impl<R> GetObjectStreamingReader<R> {
             {
                 return "read_quorum";
             }
-            if error_msg.contains("connection reset")
-                || error_msg.contains("broken pipe")
-                || error_msg.contains("downstream")
-                || error_msg.contains("remote closed")
-            {
+            if error_msg.contains("getobject output closed:") {
                 return "downstream_closed";
             }
         }
@@ -10608,6 +10604,21 @@ mod tests {
         assert_eq!(err.kind(), std::io::ErrorKind::TimedOut);
     }
 
+    #[test]
+    fn get_object_streaming_reader_distinguishes_internal_and_output_broken_pipes() {
+        let internal = std::io::Error::from(std::io::ErrorKind::BrokenPipe);
+        assert_eq!(GetObjectStreamingReader::<std::io::Cursor<Vec<u8>>>::classify_read_error(&internal), "io");
+
+        let output_closed = std::io::Error::new(
+            std::io::ErrorKind::BrokenPipe,
+            std::io::Error::other("GetObject output closed: broken pipe"),
+        );
+        assert_eq!(
+            GetObjectStreamingReader::<std::io::Cursor<Vec<u8>>>::classify_read_error(&output_closed),
+            "downstream_closed"
+        );
+    }
+
     #[tokio::test]
     async fn put_object_body_read_timeout_guard_aborts_on_stall() {
         // Inner stream never yields and never reports EOF (a proxy that forwarded

rustfs/src/server/http.rs

@@ -2368,6 +2368,48 @@ mod tests {
         rustfs_common::set_global_local_node_name(&previous_node_name).await;
     }
 
+    /// Rolling-upgrade compatibility anchor for <https://github.com/rustfs/backlog/issues/1327>:
+    /// a legacy-only peer (constant-target signature, no v2 headers) must keep authenticating
+    /// through the real production path (`check_auth` + `RpcRequestTarget` extension), and every
+    /// such acceptance must increment the v1-fallback convergence counter exactly once. That
+    /// counter reading zero fleet-wide is the precondition for ever enabling
+    /// `RUSTFS_INTERNODE_RPC_SIGNATURE_STRICT`.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rpc_auth_accepts_legacy_only_peer_and_counts_v1_fallback() {
+        use rustfs_io_metrics::internode_metrics::global_internode_metrics;
+
+        let _ = rustfs_credentials::set_global_rpc_secret("rpc-http-test-secret".to_string());
+        let previous_node_name = rustfs_common::get_global_local_node_name().await;
+        rustfs_common::set_global_local_node_name("127.0.0.1:9000").await;
+
+        // Exactly what an old peer sends on the gRPC plane: the legacy constant-target signature
+        // and timestamp headers, nothing else.
+        let legacy_headers = storage::gen_signature_headers(TONIC_RPC_PREFIX, &Method::GET).expect("legacy headers should build");
+        let mut request = Request::new(());
+        request.metadata_mut().as_mut().extend(legacy_headers);
+        request.extensions_mut().insert(RpcRequestTarget {
+            uri: "http://127.0.0.1:9000/node_service.NodeService/Ping"
+                .parse()
+                .expect("test RPC URI should parse"),
+            method: Method::POST,
+        });
+
+        let before = global_internode_metrics().snapshot().signature_v1_fallback_total;
+        assert!(
+            check_auth(request).is_ok(),
+            "a legacy-only peer must keep authenticating during rolling upgrades"
+        );
+        let after = global_internode_metrics().snapshot().signature_v1_fallback_total;
+        assert_eq!(
+            after,
+            before + 1,
+            "an accepted legacy-only request must increment signature_v1_fallback_total exactly once"
+        );
+
+        rustfs_common::set_global_local_node_name(&previous_node_name).await;
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn peer_rest_heal_control_uses_production_auth_and_keeps_validation_errors_online() {

rustfs/src/storage/storage_api.rs

@@ -488,7 +488,7 @@ pub(crate) mod ecstore_rpc {
     };
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::rpc::{
-        gen_tonic_signature_headers, set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof,
+        gen_signature_headers, gen_tonic_signature_headers, set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof,
     };
 }
 
@@ -552,6 +552,8 @@ pub(crate) fn try_current_local_node_name() -> Option<String> {
     crate::storage::runtime_sources::try_current_local_node_name()
 }
 
+#[cfg(test)]
+pub(crate) use ecstore_rpc::gen_signature_headers;
 #[cfg(test)]
 pub(crate) use ecstore_rpc::gen_tonic_signature_headers;
 pub(crate) use ecstore_rpc::sign_tonic_rpc_response_proof;

rustfs/src/storage_api.rs

@@ -106,7 +106,8 @@ pub(crate) mod server {
 
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::{
-            Endpoint, EndpointServerPools, Endpoints, PeerRestClient, PoolEndpoints, gen_tonic_signature_headers,
+            Endpoint, EndpointServerPools, Endpoints, PeerRestClient, PoolEndpoints, gen_signature_headers,
+            gen_tonic_signature_headers,
         };
 
         pub(crate) mod ecfs {