wip

Author: lcian

Cargo.lock

@@ -13,7 +13,7 @@ use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_service::{DeleteResult, GetResult, InsertResult};
 
 use crate::auth::AuthAwareService;
-use crate::error::ApiResult;
+use crate::endpoints::common::ApiResult;
 use crate::extractors::Operation;
 use crate::extractors::{BatchRequest, Xt};
 use crate::multipart::{IntoBytesStream, Part};
@@ -177,11 +177,14 @@ impl IntoPart for DeleteResult {
 
 #[cfg(test)]
 mod tests {
+    use crate::auth::PublicKeyDirectory;
+
     use super::*;
     use axum::body::Body;
     use axum::http::{Request, StatusCode};
     use bytes::Bytes;
     use objectstore_service::StorageConfig;
+    use std::collections::BTreeMap;
     use std::sync::Arc;
     use tower::ServiceExt;
 
@@ -201,6 +204,9 @@ mod tests {
         let state = Arc::new(crate::state::Services {
             config: crate::config::Config::default(),
             service: storage_service,
+            key_directory: PublicKeyDirectory {
+                keys: BTreeMap::new(),
+            },
         });
 
         // Build the router with state

objectstore-server/src/endpoints/batch.rs

@@ -4,7 +4,7 @@ use anyhow::Context;
 use axum::{
     extract::{FromRequest, Request},
     http::StatusCode,
-    response::IntoResponse,
+    response::{IntoResponse, Response},
 };
 use bytes::Bytes;
 use futures::{StreamExt, stream::BoxStream};
@@ -14,8 +14,6 @@ use multer::{Constraints, Multipart, SizeLimit};
 use objectstore_service::id::ObjectKey;
 use objectstore_types::Metadata;
 
-use crate::error::AnyhowResponse;
-
 #[derive(Debug)]
 pub struct GetOperation {
     pub key: ObjectKey,
@@ -95,37 +93,37 @@ impl<S> FromRequest<S> for BatchRequest
 where
     S: Send + Sync,
 {
-    type Rejection = AnyhowResponse;
+    type Rejection = Response;
 
     async fn from_request(request: Request, _: &S) -> Result<Self, Self::Rejection> {
         let Some(content_type) = request
             .headers()
             .get(CONTENT_TYPE)
             .and_then(|ct| ct.to_str().ok())
         else {
-            return Err((StatusCode::BAD_REQUEST, "expected valid Content-Type")
-                .into_response()
-                .into());
+            return Err((StatusCode::BAD_REQUEST, "expected valid Content-Type").into_response());
         };
 
         let Ok(mime) = content_type.parse::<mime::Mime>() else {
-            return Err((StatusCode::BAD_REQUEST, "expected valid Content-Type")
-                .into_response()
-                .into());
+            return Err((StatusCode::BAD_REQUEST, "expected valid Content-Type").into_response());
         };
         if !(mime.type_() == mime::MULTIPART && mime.subtype() == "mixed") {
             return Err((
                 StatusCode::BAD_REQUEST,
                 "expected Content-Type: multipart/mixed",
             )
-                .into_response()
-                .into());
+                .into_response());
         }
 
         // XXX: `multer::parse_boundary` requires the content-type to be `multipart/form-data`
         let content_type = content_type.replace("multipart/mixed", "multipart/form-data");
-        let boundary =
-            multer::parse_boundary(content_type).context("failed to parse multipart boundary")?;
+        let Ok(boundary) = multer::parse_boundary(content_type).context("") else {
+            return Err((
+                StatusCode::BAD_REQUEST,
+                "failed to parse multipart boundary",
+            )
+                .into_response());
+        };
         let mut parts = Multipart::with_constraints(
             request.into_body().into_data_stream(),
             boundary,

objectstore-server/src/extractors/batch.rs

@@ -80,29 +80,29 @@ where
         let mut this = self.project();
         match std::mem::replace(this.state, State::Waiting) {
             State::Waiting => match this.parts.as_mut().poll_next(ctx) {
-                Poll::Pending => return Poll::Pending,
+                Poll::Pending => Poll::Pending,
                 Poll::Ready(None) => {
                     *this.state = State::SendClosingBoundary;
                     ctx.waker().wake_by_ref();
-                    return Poll::Pending;
+                    Poll::Pending
                 }
-                Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))),
+                Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
                 Poll::Ready(Some(Ok(p))) => {
                     *this.state = State::SendHeaders(p);
-                    return Poll::Ready(Some(Ok(this.boundary.clone())));
+                    Poll::Ready(Some(Ok(this.boundary.clone())))
                 }
             },
             State::SendHeaders(part) => {
                 *this.state = State::SendBody(part.body);
                 let headers = serialize_headers(part.headers);
-                return Poll::Ready(Some(Ok(headers)));
+                Poll::Ready(Some(Ok(headers)))
             }
             State::SendBody(body) => {
                 // Add \r\n after the body
                 let mut body_with_newline = BytesMut::with_capacity(body.len() + 2);
                 body_with_newline.put(body);
                 body_with_newline.put(&b"\r\n"[..]);
-                return Poll::Ready(Some(Ok(body_with_newline.freeze())));
+                Poll::Ready(Some(Ok(body_with_newline.freeze())))
             }
             State::SendClosingBoundary => {
                 *this.state = State::Done;
@@ -114,10 +114,10 @@ where
                 let mut closing = BytesMut::with_capacity(boundary_without_crlf.len() + 4);
                 closing.put(boundary_without_crlf.as_bytes());
                 closing.put(&b"--\r\n"[..]);
-                return Poll::Ready(Some(Ok(closing.freeze())));
+                Poll::Ready(Some(Ok(closing.freeze())))
             }
             State::Done => {
-                return Poll::Ready(None);
+                Poll::Ready(None)
             }
         }
     }

objectstore-server/src/multipart.rs

@@ -10,18 +10,16 @@ mod backend;
 pub mod id;
 
 use std::path::Path;
-use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Instant;
 
 use bytes::{Bytes, BytesMut};
-use futures_util::Stream;
 use futures_util::{StreamExt, TryStreamExt, stream::BoxStream};
 use objectstore_types::Metadata;
 
 use crate::backend::common::BoxedBackend;
-use crate::id::{ObjectContext, ObjectId, ObjectKey};
+use crate::id::{ObjectContext, ObjectId};
 
 /// The threshold up until which we will go to the "high volume" backend.
 const BACKEND_SIZE_THRESHOLD: usize = 1024 * 1024; // 1 MiB