From 7e9ccf143c1537c055b933dabdadbd3a45cb8a29 Mon Sep 17 00:00:00 2001 From: RanaPriyansh Date: Sun, 20 Sep 2026 13:37:15 +0530 Subject: [PATCH] fix: accept complete HTTP responses for full ranges --- src/client/get.rs | 291 +++++++++++++++++++++++++++++++++++++++++--- src/http/client.rs | 27 +++-- src/http/mod.rs | 295 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 589 insertions(+), 24 deletions(-) diff --git a/src/client/get.rs b/src/client/get.rs index c7ec36d7..d6885b64 100644 --- a/src/client/get.rs +++ b/src/client/get.rs @@ -245,6 +245,20 @@ impl GetContext { return Err(Self::err(e)); } + if parts.status == StatusCode::OK { + let meta = header_meta( + &ctx.location, + &parts.headers, + T::HEADER_CONFIG, + ) + .map_err(Self::err)?; + if range.start >= range.end || range != (0..meta.size) { + return Err(Self::err(GetResultError::NotPartial)); + } + body = retry_body; + continue; + } + // Validate the Content-Range of the retry response let content_range = parse_range(&parts.headers).map_err(Self::err)?; @@ -313,22 +327,29 @@ fn get_range_meta( ) -> Result<(Range, ObjectMeta), GetResultError> { let mut meta = header_meta(location, &response.headers, cfg)?; let range = if let Some(expected) = range { - if response.status != StatusCode::PARTIAL_CONTENT { - return Err(GetResultError::NotPartial); - } + if response.status == StatusCode::OK { + if !is_full_representation(expected, meta.size) { + return Err(GetResultError::NotPartial); + } + 0..meta.size + } else { + if response.status != StatusCode::PARTIAL_CONTENT { + return Err(GetResultError::NotPartial); + } - let value = parse_range(&response.headers)?; - let actual = value.range; + let value = parse_range(&response.headers)?; + let actual = value.range; - // Update size to reflect the full size of the object (#5272) - meta.size = value.size; + // Update size to reflect the full size of the object (#5272) + meta.size = value.size; - let expected = expected.as_range(meta.size)?; - if actual != expected { - return Err(GetResultError::UnexpectedRange { expected, actual }); - } + let expected = expected.as_range(meta.size)?; + if actual != expected { + return Err(GetResultError::UnexpectedRange { expected, actual }); + } - actual + actual + } } else { 0..meta.size }; @@ -336,6 +357,11 @@ fn get_range_meta( Ok((range, meta)) } +pub(crate) fn is_full_representation(range: &GetRange, size: u64) -> bool { + !matches!(range, GetRange::Suffix(0)) + && range.as_range(size).is_ok_and(|range| range == (0..size)) +} + /// Extracts the [CONTENT_RANGE] header fn parse_range(headers: &http::HeaderMap) -> Result { let val = headers @@ -525,17 +551,25 @@ mod tests { } #[cfg(all(test, feature = "http-base", not(target_arch = "wasm32")))] mod http_tests { + const BODY: &str = "hello world!"; + + use super::{GetClient, GetClientExt}; + use crate::client::header::HeaderConfig; use crate::client::mock_server::MockServer; - use crate::client::{HttpError, HttpErrorKind, HttpResponseBody}; + use crate::client::retry::RetryContext; + use crate::client::{HttpError, HttpErrorKind, HttpResponse, HttpResponseBody}; use crate::http::HttpBuilder; use crate::path::Path; - use crate::{ClientOptions, ObjectStoreExt, RetryConfig}; + use crate::{ClientOptions, GetOptions, GetRange, ObjectStoreExt, RetryConfig}; + use async_trait::async_trait; use bytes::Bytes; use futures_util::FutureExt; use http::header::{CONNECTION, CONTENT_LENGTH, CONTENT_RANGE, ETAG, RANGE}; use http::{Response, StatusCode}; use hyper::body::Frame; + use std::collections::VecDeque; use std::pin::Pin; + use std::sync::{Arc, Mutex}; use std::task::{Context, Poll, ready}; use std::time::Duration; @@ -590,6 +624,235 @@ mod http_tests { } } + #[cfg(feature = "reqwest")] + struct SequenceClient { + responses: Mutex>, + ranges: Arc>>>, + retry_config: RetryConfig, + } + + #[cfg(feature = "reqwest")] + #[async_trait] + impl GetClient for SequenceClient { + const STORE: &'static str = "TEST"; + const HEADER_CONFIG: HeaderConfig = HeaderConfig { + etag_required: false, + last_modified_required: false, + version_header: None, + user_defined_metadata_prefix: None, + }; + + fn retry_config(&self) -> &RetryConfig { + &self.retry_config + } + + async fn get_request( + &self, + _ctx: &mut RetryContext, + _path: &Path, + options: GetOptions, + ) -> crate::Result { + self.ranges.lock().unwrap().push(options.range); + Ok(self.responses.lock().unwrap().pop_front().unwrap()) + } + } + + #[cfg(feature = "reqwest")] + #[tokio::test] + async fn test_retry_rejects_full_200_after_partial_body() { + let first = Response::builder() + .status(StatusCode::PARTIAL_CONTENT) + .header(CONTENT_LENGTH, 5) + .header(CONTENT_RANGE, "bytes 2-6/12") + .header(ETAG, "123") + .body(Chunked::new(vec![Ok(Bytes::from_static(b"llo")), Err(())])) + .unwrap() + .map(Into::into); + let second = Response::builder() + .header(CONTENT_LENGTH, 12) + .header(ETAG, "123") + .body(BODY.to_string()) + .unwrap() + .map(Into::into); + let ranges = Arc::new(Mutex::new(Vec::new())); + let client = Arc::new(SequenceClient { + responses: Mutex::new(VecDeque::from([first, second])), + ranges: Arc::clone(&ranges), + retry_config: RetryConfig { + backoff: Default::default(), + max_retries: 1, + retry_timeout: Duration::from_secs(1000), + }, + }); + + let result = client + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(2..7)), + ) + .await + .unwrap(); + let error = result.bytes().await.unwrap_err(); + assert!(error.to_string().contains("Received non-partial response")); + assert_eq!( + ranges.lock().unwrap().as_slice(), + [Some(GetRange::Bounded(2..7)), Some(GetRange::Bounded(5..7)),] + ); + } + + #[cfg(feature = "reqwest")] + #[tokio::test] + async fn test_retry_rejects_shorter_complete_200() { + let first = Response::builder() + .header(CONTENT_LENGTH, 12) + .header(ETAG, "123") + .body(Chunked::new(vec![Err(())])) + .unwrap() + .map(Into::into); + let second = Response::builder() + .header(CONTENT_LENGTH, 8) + .header(ETAG, "123") + .body("shorter!".to_string()) + .unwrap() + .map(Into::into); + let client = Arc::new(SequenceClient { + responses: Mutex::new(VecDeque::from([first, second])), + ranges: Arc::new(Mutex::new(Vec::new())), + retry_config: RetryConfig { + backoff: Default::default(), + max_retries: 1, + retry_timeout: Duration::from_secs(1000), + }, + }); + + let result = client + .get_opts(&Path::from("test"), GetOptions::new()) + .await + .unwrap(); + let error = result.bytes().await.unwrap_err(); + assert!(error.to_string().contains("Received non-partial response")); + } + + #[cfg(feature = "reqwest")] + #[tokio::test] + async fn test_retry_accepts_complete_200_after_empty_body() { + let first = Response::builder() + .header(CONTENT_LENGTH, 12) + .header(ETAG, "123") + .body(Chunked::new(vec![Err(())])) + .unwrap() + .map(Into::into); + let second = Response::builder() + .header(CONTENT_LENGTH, 12) + .header(CONTENT_RANGE, "bytes 8-9/10") + .header(ETAG, "123") + .body(BODY.to_string()) + .unwrap() + .map(Into::into); + let ranges = Arc::new(Mutex::new(Vec::new())); + let client = Arc::new(SequenceClient { + responses: Mutex::new(VecDeque::from([first, second])), + ranges: Arc::clone(&ranges), + retry_config: RetryConfig { + backoff: Default::default(), + max_retries: 1, + retry_timeout: Duration::from_secs(1000), + }, + }); + + let result = client + .get_opts(&Path::from("test"), GetOptions::new()) + .await + .unwrap(); + assert_eq!(result.range, 0..12); + assert_eq!(result.meta.size, 12); + assert_eq!(result.bytes().await.unwrap().as_ref(), BODY.as_bytes()); + assert_eq!( + ranges.lock().unwrap().as_slice(), + [None, Some(GetRange::Bounded(0..12))] + ); + } + + #[cfg(feature = "reqwest")] + #[tokio::test] + async fn test_retry_with_complete_200_after_empty_body() { + let mock = MockServer::new().await; + let retry = RetryConfig { + backoff: Default::default(), + max_retries: 1, + retry_timeout: Duration::from_secs(1000), + }; + let options = ClientOptions::new().with_allow_http(true); + let store = HttpBuilder::new() + .with_client_options(options) + .with_retry(retry) + .with_url(mock.url()) + .build() + .unwrap(); + let path = Path::from("test"); + + mock.push( + Response::builder() + .header(CONTENT_LENGTH, 12) + .header(ETAG, "123") + .body(Chunked::new(vec![Err(())])) + .unwrap(), + ); + mock.push( + Response::builder() + .header(CONTENT_LENGTH, 12) + .header(ETAG, "123") + .body(BODY.to_string()) + .unwrap(), + ); + + let bytes = store.get(&path).await.unwrap().bytes().await.unwrap(); + assert_eq!(bytes.as_ref(), BODY.as_bytes()); + + mock.shutdown().await; + } + + #[cfg(feature = "reqwest")] + #[tokio::test] + async fn test_retry_rejects_proper_subset_200() { + let mock = MockServer::new().await; + let retry = RetryConfig { + backoff: Default::default(), + max_retries: 1, + retry_timeout: Duration::from_secs(1000), + }; + let options = ClientOptions::new().with_allow_http(true); + let store = HttpBuilder::new() + .with_client_options(options) + .with_retry(retry) + .with_url(mock.url()) + .build() + .unwrap(); + let path = Path::from("test"); + + mock.push( + Response::builder() + .status(StatusCode::PARTIAL_CONTENT) + .header(CONTENT_LENGTH, 5) + .header(CONTENT_RANGE, "bytes 2-6/12") + .header(ETAG, "123") + .body(Chunked::new(vec![Err(())])) + .unwrap(), + ); + mock.push( + Response::builder() + .header(CONTENT_LENGTH, 12) + .header(ETAG, "123") + .body(BODY.to_string()) + .unwrap(), + ); + + let error = store.get_range(&path, 2..7).await.unwrap_err(); + assert!(matches!(error, crate::Error::NotSupported { .. })); + + mock.shutdown().await; + } + #[cfg(feature = "reqwest")] #[tokio::test] async fn test_stream_retry() { diff --git a/src/http/client.rs b/src/http/client.rs index 580b7258..a7ae9516 100644 --- a/src/http/client.rs +++ b/src/http/client.rs @@ -16,7 +16,7 @@ // under the License. use super::STORE; -use crate::client::get::GetClient; +use crate::client::get::{GetClient, is_full_representation}; use crate::client::header::HeaderConfig; use crate::client::retry::{self, RetryConfig, RetryContext, RetryExt}; use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpResponse}; @@ -378,7 +378,7 @@ impl GetClient for Client { true => Method::HEAD, false => Method::GET, }; - let has_range = options.range.is_some(); + let requested_range = options.range.clone(); let builder = self.client.request(method, url); let res = builder @@ -401,14 +401,21 @@ impl GetClient for Client { .into(), })?; - // We expect a 206 Partial Content response if a range was requested - // a 200 OK response would indicate the server did not fulfill the request - if has_range && res.status() != StatusCode::PARTIAL_CONTENT { - return Err(crate::Error::NotSupported { - source: Box::new(Error::RangeNotSupported { - href: path.to_string(), - }), - }); + if let Some(range) = requested_range { + let is_full = if res.status() == StatusCode::OK { + crate::client::header::header_meta(path, res.headers(), Self::HEADER_CONFIG) + .map(|meta| is_full_representation(&range, meta.size)) + .unwrap_or(false) + } else { + false + }; + if res.status() != StatusCode::PARTIAL_CONTENT && !is_full { + return Err(crate::Error::NotSupported { + source: Box::new(Error::RangeNotSupported { + href: path.to_string(), + }), + }); + } } Ok(res) diff --git a/src/http/mod.rs b/src/http/mod.rs index 1c83d45b..c2a18e18 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -325,3 +325,298 @@ mod tests { response_extensions(&integration, false).await; } } + +#[cfg(all( + test, + feature = "http-base", + feature = "reqwest", + not(target_arch = "wasm32") +))] +mod range_tests { + use super::HttpBuilder; + use crate::client::mock_server::MockServer; + use crate::{GetOptions, ObjectStore, Path}; + use http::header::{CONTENT_LENGTH, CONTENT_RANGE, RANGE}; + use http::{Response, StatusCode}; + + const BODY: &str = "hello world!"; + + async fn store_with_response( + response: Response, + expected_range: Option<&str>, + ) -> (MockServer, super::HttpStore) { + let mock = MockServer::new().await; + let expected_range = expected_range.map(str::to_string); + mock.push_fn(move |request| { + assert_eq!(request.uri().path(), "/test"); + match expected_range { + Some(expected) => assert_eq!( + request + .headers() + .get(RANGE) + .and_then(|value| value.to_str().ok()), + Some(expected.as_str()) + ), + None => assert!(request.headers().get(RANGE).is_none()), + } + response + }); + let store = HttpBuilder::new() + .with_url(mock.url()) + .with_client_options(crate::ClientOptions::new().with_allow_http(true)) + .build() + .unwrap(); + (mock, store) + } + + #[tokio::test] + async fn full_range_200_without_content_range_succeeds() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=0-11")).await; + + let result = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(0..BODY.len() as u64)), + ) + .await + .unwrap(); + assert_eq!(result.range, 0..BODY.len() as u64); + assert_eq!(result.meta.size, BODY.len() as u64); + assert_eq!(result.bytes().await.unwrap().as_ref(), BODY.as_bytes()); + + mock.shutdown().await; + } + + #[tokio::test] + async fn full_range_200_with_content_range_succeeds() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .header(CONTENT_RANGE, "bytes 0-11/12") + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=0-11")).await; + + let result = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(0..BODY.len() as u64)), + ) + .await + .unwrap(); + assert_eq!(result.range, 0..BODY.len() as u64); + assert_eq!(result.meta.size, BODY.len() as u64); + assert_eq!(result.bytes().await.unwrap().as_ref(), BODY.as_bytes()); + + mock.shutdown().await; + } + + #[tokio::test] + async fn full_offset_200_succeeds() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=0-")).await; + + let result = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(crate::GetRange::Offset(0))), + ) + .await + .unwrap(); + assert_eq!(result.range, 0..BODY.len() as u64); + assert_eq!(result.meta.size, BODY.len() as u64); + assert_eq!(result.bytes().await.unwrap().as_ref(), BODY.as_bytes()); + + mock.shutdown().await; + } + + #[tokio::test] + async fn full_suffix_200_succeeds() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=-12")).await; + + let result = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(crate::GetRange::Suffix(12))), + ) + .await + .unwrap(); + assert_eq!(result.range, 0..BODY.len() as u64); + assert_eq!(result.meta.size, BODY.len() as u64); + assert_eq!(result.bytes().await.unwrap().as_ref(), BODY.as_bytes()); + + mock.shutdown().await; + } + + #[tokio::test] + async fn positive_suffix_empty_200_succeeds() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, 0) + .body(String::new()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=-1")).await; + + let result = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(crate::GetRange::Suffix(1))), + ) + .await + .unwrap(); + assert_eq!(result.range, 0..0); + assert_eq!(result.meta.size, 0); + assert!(result.bytes().await.unwrap().is_empty()); + + mock.shutdown().await; + } + + #[tokio::test] + async fn misleading_content_range_200_is_ignored() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .header(CONTENT_RANGE, "bytes 9-9/999") + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=0-11")).await; + + let result = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(0..BODY.len() as u64)), + ) + .await + .unwrap(); + assert_eq!(result.range, 0..BODY.len() as u64); + assert_eq!(result.meta.size, BODY.len() as u64); + assert_eq!(result.bytes().await.unwrap().as_ref(), BODY.as_bytes()); + + mock.shutdown().await; + } + + #[tokio::test] + async fn suffix_zero_200_is_rejected() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=-0")).await; + + let error = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(crate::GetRange::Suffix(0))), + ) + .await + .unwrap_err(); + assert!(matches!(error, crate::Error::NotSupported { .. })); + + mock.shutdown().await; + } + + #[tokio::test] + async fn zero_length_bounded_200_is_rejected() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, None).await; + + let error = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(0..0)), + ) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("Range started at 0 and ended at 0") + ); + + mock.shutdown().await; + } + + #[tokio::test] + async fn proper_subset_200_remains_an_error() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=2-6")).await; + + let error = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(2..7)), + ) + .await + .unwrap_err(); + assert!(matches!(error, crate::Error::NotSupported { .. })); + + mock.shutdown().await; + } + + #[tokio::test] + async fn partial_range_206_succeeds() { + let response = Response::builder() + .status(StatusCode::PARTIAL_CONTENT) + .header(CONTENT_LENGTH, 5) + .header(CONTENT_RANGE, "bytes 2-6/12") + .body("llo w".to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, Some("bytes=2-6")).await; + + let result = store + .get_opts( + &Path::from("test"), + GetOptions::new().with_range(Some(2..7)), + ) + .await + .unwrap(); + assert_eq!(result.range, 2..7); + assert_eq!(result.meta.size, BODY.len() as u64); + assert_eq!(result.bytes().await.unwrap().as_ref(), b"llo w"); + + mock.shutdown().await; + } + + #[tokio::test] + async fn unrequested_200_succeeds() { + let response = Response::builder() + .status(StatusCode::OK) + .header(CONTENT_LENGTH, BODY.len()) + .body(BODY.to_string()) + .unwrap(); + let (mock, store) = store_with_response(response, None).await; + + let result = store + .get_opts(&Path::from("test"), GetOptions::new()) + .await + .unwrap(); + assert_eq!(result.range, 0..BODY.len() as u64); + assert_eq!(result.meta.size, BODY.len() as u64); + assert_eq!(result.bytes().await.unwrap().as_ref(), BODY.as_bytes()); + + mock.shutdown().await; + } +}