fix: resume GET body after peer short EOF (#7852)

* fix: resume get body after short eof

Treat mid-stream short EOF errors as resumable when the S3 GET body has an attached resume context. This lets peer-kill reads reopen the same object version at the emitted offset instead of aborting the 200 response body.

Co-Authored-By: heihutu <heihutu@gmail.com>

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

* fix(connect): surface TLS peer bootstrap failures

Map Connect registration TLS peer validation failures to the dedicated bootstrap error so wrong CA inputs fail closed with the sanitized TLS variant instead of the generic exchange error.

Co-Authored-By: heihutu <heihutu@gmail.com>

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

---------

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

rustfs/src/app/object/get.rs

@@ -1632,12 +1632,15 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                     return Poll::Ready(Ok(()));
                 }
                 Poll::Ready(Err(err)) => {
-                    // Typed relocation errors (the codec read path delivers them
-                    // in-band) mean rebalance/decommission removed the pinned
-                    // object data mid-stream: reopen and continue instead of
-                    // failing the download. The error is only intercepted before
-                    // the committed body length has been fully delivered.
-                    if self.emitted < self.expected && is_object_relocation_error(&err) && self.resume.is_some() {
+                    // Typed relocation errors and bounded short EOF errors both
+                    // mean the committed body stopped before Content-Length was
+                    // delivered. Reopen the same object/version at the emitted
+                    // offset when resume is attached; if reopen cannot prove the
+                    // same identity, it fails closed with this trigger error.
+                    if self.emitted < self.expected
+                        && (is_object_relocation_error(&err) || is_resumable_short_eof_error(&err))
+                        && self.resume.is_some()
+                    {
                         self.begin_resume(err);
                         continue;
                     }
@@ -1956,6 +1959,13 @@ fn is_object_relocation_error(err: &std::io::Error) -> bool {
     }
 }
 
+fn is_resumable_short_eof_error(err: &std::io::Error) -> bool {
+    err.kind() == std::io::ErrorKind::UnexpectedEof
+        || err
+            .get_ref()
+            .is_some_and(|source| source.downcast_ref::<rustfs_rio::IncompleteBody>().is_some())
+}
+
 pub(crate) fn object_seek_support_threshold() -> usize {
     static OBJECT_SEEK_SUPPORT_THRESHOLD: OnceLock<usize> = OnceLock::new();
     *OBJECT_SEEK_SUPPORT_THRESHOLD.get_or_init(|| {
@@ -7919,6 +7929,46 @@ mod tests {
         assert_eq!(reopen_count.load(Ordering::Relaxed), 1);
     }
 
+    #[tokio::test]
+    async fn get_object_streaming_reader_resumes_after_short_eof_error() {
+        use tokio::io::AsyncReadExt;
+
+        // The remote-disk path can retire a peer stream with an UnexpectedEof
+        // once its bounded fresh-open recovery is exhausted. The S3 body still
+        // has enough identity to reopen the object and continue from the bytes
+        // already sent to the client.
+        let reopen_count = Arc::new(AtomicUsize::new(0));
+        let control = counting_resume_control(Arc::clone(&reopen_count), |emitted| {
+            assert_eq!(emitted, 6, "resume must reopen at the emitted offset");
+            Ok(FailAtEndReader::new(b"world", None))
+        });
+        let mut reader = GetObjectStreamingReader::new(
+            FailAtEndReader::new(
+                b"hello ",
+                Some(std::io::Error::new(
+                    std::io::ErrorKind::UnexpectedEof,
+                    "remote read ended before requested length",
+                )),
+            ),
+            "test-bucket",
+            "remote-eof-object",
+            "req-resume-remote-eof",
+            None,
+            11,
+            Duration::ZERO,
+            GetObjectBodyLifecycle::disabled(),
+            Some(control),
+        );
+        let mut out = Vec::new();
+        reader
+            .read_to_end(&mut out)
+            .await
+            .expect("a short EOF from a retired remote shard must resume the committed body");
+
+        assert_eq!(out, b"hello world");
+        assert_eq!(reopen_count.load(Ordering::Relaxed), 1);
+    }
+
     #[tokio::test]
     async fn get_object_streaming_reader_clean_eof_does_not_resume() {
         use tokio::io::AsyncReadExt;

rustfs/src/connect/registration_bootstrap.rs

@@ -120,6 +120,7 @@ pub async fn register_from_protected_input(
         .map_err(|error| match error {
             super::ClientError::ProxyAuthentication => RegistrationBootstrapError::ProxyAuthentication,
             super::ClientError::ProxyRejected => RegistrationBootstrapError::ProxyRejected,
+            super::ClientError::TlsPeer => RegistrationBootstrapError::TlsPeer,
             _ => RegistrationBootstrapError::Exchange,
         })?;
     if credential.name != format!("{cluster_name}/clusterDevices/{}", credential.uid) {

rustfs/tests/connect_registration_bootstrap.rs

@@ -483,7 +483,7 @@ async fn endpoint_ca_token_state_and_service_failures_are_closed_and_sanitized()
     let error = register_from_protected_input(&server.endpoint, &wrong_root, &wrong_ca_state, Some(&token))
         .await
         .expect_err("wrong CA must fail TLS");
-    assert!(matches!(error, RegistrationBootstrapError::Exchange));
+    assert!(matches!(error, RegistrationBootstrapError::TlsPeer));
     assert!(!wrong_ca_state.join("credential/device.crt.json").exists());
 }