diff --git a/crates/config/src/constants/internode.rs b/crates/config/src/constants/internode.rs index 8814b8b3e..aecbb10cf 100644 --- a/crates/config/src/constants/internode.rs +++ b/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 (). +/// +/// 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); diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 1b1802795..1c387160e 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/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 = 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 +/// . +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(); diff --git a/crates/ecstore/src/diagnostics/get.rs b/crates/ecstore/src/diagnostics/get.rs index b72b16cfb..cde5a5536 100644 --- a/crates/ecstore/src/diagnostics/get.rs +++ b/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::() { + 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"); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 8ba420268..47de63c8c 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 5136cde92..8eb4b590f 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index b43b9c70f..b0e2bab5a 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/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 AsyncWrite for GetObjectOutputWriter<'_, W> { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + 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> { + 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> { + 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, disks: &[Option], @@ -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> { + Poll::Ready(Err(std::io::Error::from(ErrorKind::BrokenPipe))) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + 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, Vec>) { 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, diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index d620e17b5..2c0985a56 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/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 + /// ). 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. diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 776114691..f91ff09fb 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1500,11 +1500,7 @@ impl GetObjectStreamingReader { { 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::>>::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::>>::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 diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index d072c2c9e..895f43378 100644 --- a/rustfs/src/server/http.rs +++ b/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 : + /// 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() { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 91fb0430e..da959a502 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/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 { 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; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index fa2bed37f..13c4e638a 100644 --- a/rustfs/src/storage_api.rs +++ b/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 {