feat: Initial support for HTTP range requests (#481)

Adds standard HTTP [range
requests](https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Range_requests)
([RFC](https://www.rfc-editor.org/rfc/rfc9110.html#name-range-requests))
support for single-object GET operations.

Spec vs implementation: 
- This implementation only supports byte ranges. The spec doesn't
specifically talk about bytes, so in theory it admits ranges of other
"types", which just don't make sense for us.
- This implementation only supports [Single part
ranges](https://developer.mozilla.org/enUS/docs/Web/HTTP/Guides/Range_requests#single_part_ranges),
meaning stuff like `Range: bytes=0-1023`. The spec allows for [Multipart
ranges](https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Range_requests#multipart_ranges)
too, i.e. fetching multiple ranges in a single request, which we don't
support. We can add it in the future if needed.
- We should probably implement [Conditional range
requests](https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Range_requests#conditional_range_requests)
in a follow-up PR, as they're pretty much essential to guarantee that
clients read consistent data (if the object is swapped between range
reads, we would otherwise serve chunks from 2 potentially different
objects). Note that Symbolicator doesn't use these, so we should also
update Symbolicator and any other user that uses range reads to actually
use these to prevent random failures/data inconsistency. (maybe in
practice this is fine as data that Symbolicator fetches cannot really
change due to the way it's written, I don't know yet)
**Looking for feedback on whether we think we need this now or not**.

Implementation details:
- All `get_object` paths now take `Option<ByteRange>` and return
`ContentRange` alongside the payload. This seems to be the way to
implement this with the least code duplication. Otherwise we would have
to introduce a `get_object_range`, implement it for every backend and
wrappers of the backends, etc.
- Bigtable: we know that the objects there are small, so we just return
the full payload. This behavior is compliant with the spec.
- GCS/S3: `objectstore-service` reserializes the `Range` header and
forwards it to the backend. On the response path, we proxy back the
`content-range` provided by the backend. TODO:
- Not supported for batch operations.
- Not implemented in clients for now because Symbolicator (which would
be the only user atm) doesn't use our client. We can easily implement it
when needed.

Close FS-363

Author: lcian

objectstore-server/src/auth/service.rs

@@ -7,6 +7,7 @@ use objectstore_service::service::{DeleteResponse, GetResponse, InsertResponse,
 use objectstore_service::{ClientStream, StorageService};
 use objectstore_types::auth::Permission;
 use objectstore_types::metadata::Metadata;
+use objectstore_types::range::ByteRange;
 
 use crate::auth::{AuthContext, AuthError};
 use crate::endpoints::common::ApiResult;
@@ -110,9 +111,13 @@ impl AuthAwareService {
     }
 
     /// Auth-aware wrapper around [`StorageService::get_object`].
-    pub async fn get_object(&self, id: ObjectId) -> ApiResult<GetResponse> {
+    pub async fn get_object(
+        &self,
+        id: ObjectId,
+        range: Option<ByteRange>,
+    ) -> ApiResult<GetResponse> {
         self.assert_authorized(Permission::ObjectRead, id.context())?;
-        Ok(self.service.get_object(id).await?)
+        Ok(self.service.get_object(id, range).await?)
     }
 
     /// Auth-aware wrapper around [`StorageService::delete_object`].

objectstore-server/src/endpoints/batch.rs

@@ -147,7 +147,7 @@ async fn convert_to_part(
     match result {
         Ok(OpResponse::Got {
             key,
-            response: Some((metadata, stream)),
+            response: Some((metadata, _content_range, stream)),
         }) => got_to_part(idx, key, metadata, stream, state, context)
             .await
             .unwrap_or_else(|e| create_error_part(idx, &e)),

objectstore-server/src/endpoints/common.rs

@@ -5,6 +5,7 @@ use std::error::Error;
 use axum::Json;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
+use http::HeaderValue;
 use objectstore_service::error::Error as ServiceError;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
@@ -93,6 +94,9 @@ impl ApiError {
 
             ApiError::Service(ServiceError::Client(_)) => StatusCode::BAD_REQUEST,
             ApiError::Service(ServiceError::Metadata(_)) => StatusCode::BAD_REQUEST,
+            ApiError::Service(ServiceError::RangeNotSatisfiable { .. }) => {
+                StatusCode::RANGE_NOT_SATISFIABLE
+            }
             ApiError::Service(ServiceError::InvalidUploadId(_)) => StatusCode::BAD_REQUEST,
             ApiError::Service(ServiceError::AtCapacity) => StatusCode::TOO_MANY_REQUESTS,
             ApiError::Service(ServiceError::NotImplemented) => StatusCode::NOT_IMPLEMENTED,
@@ -115,3 +119,11 @@ impl IntoResponse for ApiError {
         (self.status(), Json(body)).into_response()
     }
 }
+
+/// Inserts `Accept-Ranges: bytes` into the response headers.
+pub fn insert_accept_ranges(response: &mut Response) {
+    response.headers_mut().insert(
+        http::header::ACCEPT_RANGES,
+        HeaderValue::from_static("bytes"),
+    );
+}

objectstore-server/src/endpoints/objects.rs

@@ -9,10 +9,12 @@ use axum::{Json, Router};
 use objectstore_service::error::Error as ServiceError;
 use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_types::metadata::Metadata;
+use objectstore_types::range::ContentRange;
 use serde::Serialize;
 
 use crate::auth::AuthAwareService;
-use crate::endpoints::common::{ApiError, ApiResult};
+use crate::endpoints::common::{ApiError, ApiResult, insert_accept_ranges};
+use crate::extractors::byte_range::OptionalByteRange;
 use crate::extractors::{Xt, body::MeteredBody};
 use crate::state::ServiceState;
 
@@ -64,15 +66,55 @@ async fn object_get(
     service: AuthAwareService,
     State(state): State<ServiceState>,
     Xt(id): Xt<ObjectId>,
+    OptionalByteRange(byte_range): OptionalByteRange,
+    _headers: HeaderMap,
 ) -> ApiResult<Response> {
     let context = id.context().clone();
-    let Some((metadata, stream)) = service.get_object(id).await? else {
-        return Ok(StatusCode::NOT_FOUND.into_response());
+    let result = service.get_object(id, byte_range).await;
+
+    let (metadata, content_range, stream) = match result {
+        Ok(Some(result)) => result,
+        Ok(None) => return Ok(StatusCode::NOT_FOUND.into_response()),
+        Err(ApiError::Service(ServiceError::RangeNotSatisfiable { total })) => {
+            let mut response = (
+                StatusCode::RANGE_NOT_SATISFIABLE,
+                [(
+                    http::header::CONTENT_RANGE,
+                    ContentRange::unsatisfiable_total_to_header_value(total),
+                )],
+            )
+                .into_response();
+            insert_accept_ranges(&mut response);
+            return Ok(response);
+        }
+        Err(e) => return Err(e),
     };
+
     let stream = state.meter_stream(stream, &context);
+    let metadata_headers = metadata.to_headers("").map_err(ServiceError::from)?;
+
+    let mut response = match content_range {
+        Some(ref content_range) => {
+            let mut resp = (
+                StatusCode::PARTIAL_CONTENT,
+                metadata_headers,
+                Body::from_stream(stream),
+            )
+                .into_response();
+            let headers = resp.headers_mut();
+            headers.insert(
+                http::header::CONTENT_LENGTH,
+                content_range.len_to_header_value(),
+            );
+            headers.insert(http::header::CONTENT_RANGE, content_range.to_header_value());
+            resp
+        }
+        None => (StatusCode::OK, metadata_headers, Body::from_stream(stream)).into_response(),
+    };
 
-    let headers = metadata.to_headers("").map_err(ServiceError::from)?;
-    Ok((headers, Body::from_stream(stream)).into_response())
+    insert_accept_ranges(&mut response);
+
+    Ok(response)
 }
 
 async fn object_head(service: AuthAwareService, Xt(id): Xt<ObjectId>) -> ApiResult<Response> {
@@ -82,7 +124,9 @@ async fn object_head(service: AuthAwareService, Xt(id): Xt<ObjectId>) -> ApiResu
 
     let headers = metadata.to_headers("").map_err(ServiceError::from)?;
 
-    Ok((StatusCode::NO_CONTENT, headers).into_response())
+    let mut response = (StatusCode::NO_CONTENT, headers).into_response();
+    insert_accept_ranges(&mut response);
+    Ok(response)
 }
 
 async fn object_put(

objectstore-server/src/extractors/byte_range.rs

@@ -0,0 +1,49 @@
+//! Axum extractor for range requests.
+
+use axum::extract::FromRequestParts;
+use http::request::Parts;
+use objectstore_types::range::{ByteRange, RangeError};
+
+use crate::{endpoints::common::ApiError, state::ServiceState};
+
+/// Extractor that parses the `Range` request header into an optional [`ByteRange`].
+#[derive(Debug, Clone)]
+pub struct OptionalByteRange(pub Option<ByteRange>);
+
+impl FromRequestParts<ServiceState> for OptionalByteRange {
+    type Rejection = ApiError;
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        _state: &ServiceState,
+    ) -> Result<Self, Self::Rejection> {
+        let headers = &parts.headers;
+        let Some(range) = headers.get(http::header::RANGE) else {
+            return Ok(Self(None));
+        };
+        let range = range
+            .to_str()
+            .map_err(|_| ApiError::Client("invalid Range header".into()))?;
+
+        match range.parse::<ByteRange>() {
+            Ok(range) => Ok(Self(Some(range))),
+            // Per RFC 9110:
+            // > A server that supports range requests MAY ignore or reject a Range header
+            //   field that contains an invalid ranges-specifier [...]
+            //
+            // If the client wants multiple ranges, fall back to returning the whole object.
+            // We might support multiple ranges in the future, so log a warning to let us know
+            // clients are trying to do this.
+            Err(RangeError::MultiRange) => {
+                objectstore_log::warn!(
+                    "received range request with multiple range specifiers, ignoring"
+                );
+                Ok(Self(None))
+            }
+            // The client requested an invalid unit or sent a malformed header.
+            // We could fall back, but better fail hard and let them know they sent something
+            // invalid.
+            Err(err) => Err(ApiError::Client(format!("invalid Range header: {err}"))),
+        }
+    }
+}

objectstore-server/src/extractors/mod.rs

@@ -2,6 +2,7 @@
 
 pub mod batch;
 pub mod body;
+pub mod byte_range;
 pub mod downstream_service;
 mod id;
 mod service;