feat: Initial support for HTTP range requests

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;