From 34917c7947d4092ce609eb47229a3a1543d1c43b Mon Sep 17 00:00:00 2001 From: Hauser Date: Thu, 8 Oct 2026 14:11:36 +0800 Subject: [PATCH] fix: resume stalled GET bodies and classify transient storage failures (#8414) * fix: resume stalled GET bodies Retry a stalled source through the existing identity-fenced resume budget so transient stalls do not truncate committed responses. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix: preserve data usage read quorum errors Apply the existing system-volume error mapping to snapshot body reads so a metadata-volume outage remains a retryable quorum failure. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix: handle transient heal lock failures Align erasure-set heal classification with the shared lock taxonomy and avoid duplicate error logging for retried scanner journal writes. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/ecstore/src/api/mod.rs | 9 +- crates/ecstore/src/data_usage/mod.rs | 76 +++++++++++++++- crates/heal/src/heal/erasure_healer.rs | 46 ++++++++++ rustfs/src/app/object/get.rs | 99 +++++++++++++++++++++ rustfs/src/storage/scanner_dirty_journal.rs | 13 ++- 5 files changed, 236 insertions(+), 7 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 121a81878..e45d56f09 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -362,10 +362,11 @@ pub mod config { ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, delete_config_no_lock, is_server_config_corrupt_error, lookup_configs, read_config, read_config_no_lock, read_config_with_metadata, read_config_without_migrate, read_config_without_migrate_no_lock, read_existing_server_config_no_lock, - read_server_config_snapshot, save_config, save_config_no_lock, save_config_with_opts, save_server_config, - save_server_config_no_lock, save_server_config_snapshot, save_server_config_snapshot_with_generation, - server_config_path, try_migrate_server_config, with_config_object_read_lock, with_config_object_write_lock, - with_server_config_read_lock, with_server_config_write_lock, + read_server_config_snapshot, save_config, save_config_no_lock, save_config_with_opts, save_config_with_opts_quiet, + save_server_config, save_server_config_no_lock, save_server_config_snapshot, + save_server_config_snapshot_with_generation, server_config_path, try_migrate_server_config, + with_config_object_read_lock, with_config_object_write_lock, with_server_config_read_lock, + with_server_config_write_lock, }; } diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index b1ad77e75..af803a3a2 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -738,7 +738,12 @@ where .filter(|etag| !etag.is_empty()) .map(str::to_owned) .ok_or_else(|| Error::other("data usage snapshot has no ETag"))?; - let mut data_usage_info = parse_usage_snapshot(&reader.read_all().await?)?; + let mut data_usage_info = parse_usage_snapshot( + &reader + .read_all() + .await + .map_err(|err| map_data_usage_metadata_read_error(err, object))?, + )?; populate_backward_compatible_usage_maps(&mut data_usage_info); validate_complete_usage_snapshot(&mut data_usage_info); Ok(Some((data_usage_info, revision))) @@ -2997,6 +3002,61 @@ mod tests { } } + #[derive(Debug)] + struct UsageSnapshotReadFailureStore; + + struct UsageSnapshotFailureReader; + + impl tokio::io::AsyncRead for UsageSnapshotFailureReader { + fn poll_read( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + _buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Err(std::io::Error::other(Error::VolumeNotFound))) + } + } + + #[async_trait::async_trait] + impl crate::storage_api_contracts::object::ObjectIO for UsageSnapshotReadFailureStore { + type Error = Error; + type RangeSpec = crate::storage_api_contracts::range::HTTPRangeSpec; + type HeaderMap = http::HeaderMap; + type ObjectOptions = ObjectOptions; + type ObjectInfo = ObjectInfo; + type GetObjectReader = crate::object_api::GetObjectReader; + type PutObjectReader = PutObjReader; + + async fn get_object_reader( + &self, + _bucket: &str, + _object: &str, + _range: Option, + _headers: Self::HeaderMap, + _opts: &Self::ObjectOptions, + ) -> Result { + Ok(crate::object_api::GetObjectReader { + stream: Box::new(UsageSnapshotFailureReader), + object_info: ObjectInfo { + etag: Some("usage-revision".to_string()), + ..Default::default() + }, + buffered_body: None, + body_source: Default::default(), + }) + } + + async fn put_object( + &self, + _bucket: &str, + _object: &str, + _data: &mut Self::PutObjectReader, + _opts: &Self::ObjectOptions, + ) -> Result { + unimplemented!("the read-failure fixture does not write") + } + } + /// Minimal ObjectIO backing `load_data_usage_cache` tests: records the keys /// read and fails the first N reads with a transient (non-absence) error. #[derive(Debug, Default)] @@ -3161,6 +3221,20 @@ mod tests { } } + #[tokio::test] + async fn data_usage_removal_maps_missing_system_volume_during_snapshot_stream_read() { + let store = UsageSnapshotReadFailureStore; + + let error = load_data_usage_for_bucket_removal(&store, "bucket-metadata/.usage.json") + .await + .expect_err("a system volume disappearing during body read must remain a retryable storage error"); + + assert_eq!( + error, + Error::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), "bucket-metadata/.usage.json".to_string()) + ); + } + #[tokio::test] async fn load_data_usage_cache_retries_a_transient_failure() { let name = "usage-cache"; diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 32fee9ff6..5a02978ba 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -337,6 +337,12 @@ impl ErasureSetHealer { return HealObjectOutcome::Failed; }; + if let EcstoreError::Lock(lock_error) = se + && !lock_error.is_fatal() + { + return HealObjectOutcome::Transient; + } + // Genuine object/version absence: nothing left to heal, treat as handled. if matches!( se, @@ -2142,6 +2148,46 @@ mod tests { )); } + #[test] + fn retryable_lock_failures_are_transient_but_fatal_lock_errors_are_not() { + use rustfs_lock::LockError; + + assert!(matches!( + ErasureSetHealer::classify_heal_object_error(&Error::Storage(EcstoreError::Lock(LockError::timeout( + "bucket/object", + std::time::Duration::from_secs(5) + )))), + HealObjectOutcome::Transient + )); + assert!(matches!( + ErasureSetHealer::classify_heal_object_error(&Error::Storage(EcstoreError::Lock(LockError::QuorumNotReached { + required: 3, + achieved: 1, + }))), + HealObjectOutcome::Transient + )); + assert!(matches!( + ErasureSetHealer::classify_heal_object_error(&Error::Storage(EcstoreError::Lock(LockError::already_locked( + "bucket/object", + "node-2" + )))), + HealObjectOutcome::Transient + )); + assert!(matches!( + ErasureSetHealer::classify_heal_object_error(&Error::Storage(EcstoreError::Lock(LockError::InsufficientNodes { + required: 3, + available: 1, + }))), + HealObjectOutcome::Transient + )); + assert!(matches!( + ErasureSetHealer::classify_heal_object_error(&Error::Storage(EcstoreError::Lock(LockError::permission_denied( + "lock policy" + )))), + HealObjectOutcome::Failed + )); + } + #[test] fn dangling_delete_grace_is_transient() { assert!(matches!( diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index b4ab95a06..bdcb79d26 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -1481,6 +1481,14 @@ impl GetObjectStreamingReader { state = "stall_timeout", "GetObject streaming body stalled" ); + if self.emitted < self.expected && self.resume.is_some() && !self.resume_in_flight() { + self.begin_resume(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "get object streaming body stall timeout", + )); + cx.waker().wake_by_ref(); + return Poll::Pending; + } self.finish_err(); return Poll::Ready(Err(std::io::Error::new( std::io::ErrorKind::TimedOut, @@ -7796,6 +7804,40 @@ mod tests { } } + struct PendingAfterReader { + data: std::io::Cursor>, + pending_after_data: bool, + } + + impl PendingAfterReader { + fn new(data: &[u8], pending_after_data: bool) -> Self { + Self { + data: std::io::Cursor::new(data.to_vec()), + pending_after_data, + } + } + } + + impl AsyncRead for PendingAfterReader { + fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let position = usize::try_from(self.data.position()).unwrap_or(usize::MAX); + let source_len = self.data.get_ref().len(); + if position >= source_len { + if self.pending_after_data { + return Poll::Pending; + } + return Poll::Ready(Ok(())); + } + let want = buf.remaining().min(source_len - position); + if want == 0 { + return Poll::Ready(Ok(())); + } + buf.put_slice(&self.data.get_ref()[position..position + want]); + self.data.set_position(u64::try_from(position + want).unwrap_or(u64::MAX)); + Poll::Ready(Ok(())) + } + } + fn relocation_read_error() -> std::io::Error { std::io::Error::other(StorageError::FileNotFound) } @@ -8006,6 +8048,63 @@ mod tests { } } + #[tokio::test(start_paused = true)] + async fn get_object_streaming_reader_resumes_after_body_stall() { + use tokio::io::AsyncReadExt; + + let reopen_count = Arc::new(AtomicUsize::new(0)); + let reopen: GetObjectReopen = Box::new({ + let reopen_count = Arc::clone(&reopen_count); + move |emitted| { + assert_eq!(emitted, 6, "resume must continue at the delivered byte offset"); + reopen_count.fetch_add(1, Ordering::Relaxed); + Box::pin(async { Ok(PendingAfterReader::new(b"world", false)) }) + } + }); + let control = GetObjectResumeControl::new( + reopen, + RetryTimer::new( + GET_OBJECT_RESUME_MAX_ATTEMPTS, + Duration::from_millis(1), + Duration::from_millis(2), + rustfs_utils::retry::NO_JITTER, + 0, + ), + ); + let mut reader = GetObjectStreamingReader::new( + PendingAfterReader::new(b"hello ", true), + "test-bucket", + "stalled-object", + "req-resume-stall", + None, + 11, + Duration::from_secs(1), + GetObjectBodyLifecycle::disabled(), + Some(control), + ); + + let mut first = [0; 6]; + assert_eq!(reader.read(&mut first).await.expect("initial bytes must be readable"), 6); + assert_eq!(&first, b"hello "); + + let mut tail = Vec::new(); + let read_to_end = reader.read_to_end(&mut tail); + tokio::pin!(read_to_end); + assert!( + futures::poll!(&mut read_to_end).is_pending(), + "the body should stall before the watchdog fires" + ); + tokio::time::advance(Duration::from_secs(1)).await; + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_millis(3)).await; + + read_to_end + .await + .expect("a timed-out source read must resume from the same object version"); + assert_eq!(tail, b"world"); + assert_eq!(reopen_count.load(Ordering::Relaxed), 1); + } + #[tokio::test(start_paused = true)] async fn get_object_streaming_reader_releases_failed_disk_permit_before_reopen() { use tokio::io::AsyncReadExt; diff --git a/rustfs/src/storage/scanner_dirty_journal.rs b/rustfs/src/storage/scanner_dirty_journal.rs index ff7d9b2e8..25f1f8373 100644 --- a/rustfs/src/storage/scanner_dirty_journal.rs +++ b/rustfs/src/storage/scanner_dirty_journal.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::{ECStore, Error, ecstore_config}; +use super::{ECStore, Error, ObjectOptions, ecstore_config}; use rustfs_scanner::{ ScannerDirtyUsageBucket, ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayRecord, ScannerDurableDirtyUsageReplayScope, SegmentInvalidationProducerIdentity, @@ -467,7 +467,16 @@ async fn flush_durable_dirty_usage_journal(store: Arc, owner: &str, sta return false; } }; - ecstore_config::com::save_config(store, &path, bytes).await + ecstore_config::com::save_config_with_opts_quiet( + store, + &path, + bytes, + &ObjectOptions { + max_parity: true, + ..Default::default() + }, + ) + .await }; match result { Ok(()) | Err(Error::ConfigNotFound) => true,