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 <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

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,
         };
     }
 

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::io::Result<()>> {
+            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<Self::RangeSpec>,
+            _headers: Self::HeaderMap,
+            _opts: &Self::ObjectOptions,
+        ) -> Result<Self::GetObjectReader, Self::Error> {
+            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<Self::ObjectInfo, Self::Error> {
+            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";

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!(

rustfs/src/app/object/get.rs

@@ -1481,6 +1481,14 @@ impl<R> GetObjectStreamingReader<R> {
                 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<Vec<u8>>,
+        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<std::io::Result<()>> {
+            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<PendingAfterReader> = 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;

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<ECStore>, 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,