From 77e2dab016344cdab8df26b1e0636738f8f03269 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Thu, 6 Aug 2026 12:19:18 +0200 Subject: [PATCH] feat(server): Add resumable upload API --- objectstore-server/docs/architecture.md | 53 +- objectstore-server/src/auth/service.rs | 60 +++ objectstore-server/src/endpoints/common.rs | 14 + objectstore-server/src/endpoints/mod.rs | 109 ++++ objectstore-server/src/endpoints/objects.rs | 51 +- objectstore-server/src/endpoints/resumable.rs | 491 ++++++++++++++++++ objectstore-server/tests/resumable.rs | 312 +++++++++++ objectstore-service/docs/architecture.md | 24 + objectstore-service/src/backend/common.rs | 109 ++++ objectstore-service/src/backend/counting.rs | 49 ++ objectstore-service/src/backend/testing.rs | 93 ++++ objectstore-service/src/backend/tiered.rs | 10 + objectstore-service/src/error.rs | 26 + objectstore-service/src/lib.rs | 1 + objectstore-service/src/resumable.rs | 46 ++ objectstore-service/src/service.rs | 224 +++++++- objectstore-types/src/lib.rs | 1 + objectstore-types/src/resumable.rs | 231 ++++++++ 18 files changed, 1856 insertions(+), 48 deletions(-) create mode 100644 objectstore-server/src/endpoints/resumable.rs create mode 100644 objectstore-server/tests/resumable.rs create mode 100644 objectstore-service/src/resumable.rs create mode 100644 objectstore-types/src/resumable.rs diff --git a/objectstore-server/docs/architecture.md b/objectstore-server/docs/architecture.md index e665fc96..6dfc4fc0 100644 --- a/objectstore-server/docs/architecture.md +++ b/objectstore-server/docs/architecture.md @@ -7,49 +7,16 @@ core storage operations. ## Endpoints -All object operations live under the `/v1/` prefix: - -| Method | Path | Description | -|----------|-------------------------------------------|------------------------------| -| `POST` | `/v1/objects/{usecase}/{scopes}/` | Insert with server-generated key | -| `GET` | `/v1/objects/{usecase}/{scopes}/{*key}` | Retrieve object | -| `HEAD` | `/v1/objects/{usecase}/{scopes}/{*key}` | Retrieve metadata only | -| `PUT` | `/v1/objects/{usecase}/{scopes}/{*key}` | Insert or overwrite with key | -| `DELETE` | `/v1/objects/{usecase}/{scopes}/{*key}` | Delete object | -| `POST` | `/v1/objects:batch/{usecase}/{scopes}/` | Batch operations (multipart) | - -### Multipart Upload Endpoints - -| Method | Path | Description | -|-----------|--------------------------------------------------------------|--------------------------------------| -| `POST` | `/v1/objects:multipart/{usecase}/{scopes}/` | Initiate upload (server-generated key) | -| `PUT` | `/v1/objects:multipart/{usecase}/{scopes}/{*key}` | Initiate upload (user-provided key) | -| `PUT` | `/v1/objects:multipart:parts/{usecase}/{scopes}/{*key}` | Upload a part (`uploadId`, `partNumber` query params) | -| `GET` | `/v1/objects:multipart:parts/{usecase}/{scopes}/{*key}` | List uploaded parts (`uploadId` query param) | -| `POST` | `/v1/objects:multipart:complete/{usecase}/{scopes}/{*key}` | Complete upload (`uploadId` query param) | -| `DELETE` | `/v1/objects:multipart/{usecase}/{scopes}/{*key}` | Abort upload (`uploadId` query param) | - -The initiate POST endpoint accepts both trailing-slash and non-trailing-slash forms. - -The complete endpoint returns `200 OK` immediately, with a streaming body that -will contain the error (if any) as JSON. Whitespace is sent in the streaming body -to keep the connection open. -Clients must parse the body to determine the actual outcome, and not rely on the -status code. - -Scopes are encoded in the URL path using Matrix URI syntax: -`org=123;project=456`. An underscore (`_`) represents empty scopes. - -### Internal Endpoints - -Internal endpoints are exempt from authentication, rate limiting, and the web -concurrency limit so they remain available when the server is under load. - -| Method | Path | Description | -|--------|------|-------------| -| `GET` | `/health` | Liveness probe (always returns 200) | -| `GET` | `/ready` | Readiness probe (returns 503 when `/tmp/objectstore.down` exists, enabling graceful drain) | -| `GET` | `/keda` | Prometheus text-format gauges for KEDA autoscaling (see [KEDA Metrics](#keda-metrics)) | +All object operations live under the `/v1/` prefix. Objects are addressed by a +usecase, a set of scopes, and a key; scopes are encoded in the URL path using +Matrix URI syntax (`org=123;project=456`). + +Four families of routes exist: object operations, resumable uploads, multipart +uploads (being replaced by resumable uploads), and internal probes that stay +available when the server is under load. + +See the [`endpoints`] module for the routing table and the request and response +shape of every route. ## Request Flow diff --git a/objectstore-server/src/auth/service.rs b/objectstore-server/src/auth/service.rs index f4643bda..cee66915 100644 --- a/objectstore-server/src/auth/service.rs +++ b/objectstore-server/src/auth/service.rs @@ -3,6 +3,9 @@ use objectstore_service::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use objectstore_service::resumable::{ + CreateSessionResponse, SessionToken, TerminateUploadResponse, UploadProgress, +}; use objectstore_service::service::{DeleteResponse, GetResponse, InsertResponse, MetadataResponse}; use objectstore_service::{ClientStream, StorageService}; @@ -186,4 +189,61 @@ impl AuthAwareService { .complete_multipart(id, upload_id, parts) .await?) } + + // --- Resumable upload operations --- + // + // Every operation requires `ObjectWrite`, including the two that do not obviously write: + // an offset query can commit an assembled object, and terminating a session discards an + // in-progress upload rather than deleting an object. So `DELETE ?session=` needs write + // permission where a plain `DELETE` on the same path needs delete permission. + + /// Auth-aware wrapper around [`StorageService::create_upload_session`]. + pub async fn create_upload_session( + &self, + id: ObjectId, + metadata: Metadata, + total_length: u64, + ) -> ApiResult { + self.check_permission(Permission::ObjectWrite, id.context())?; + Ok(self + .service + .create_upload_session(id, metadata, total_length) + .await?) + } + + /// Auth-aware wrapper around [`StorageService::put_chunk`]. + pub async fn put_chunk( + &self, + id: ObjectId, + session: SessionToken, + offset: u64, + content_length: u64, + body: ClientStream, + ) -> ApiResult { + self.check_permission(Permission::ObjectWrite, id.context())?; + Ok(self + .service + .put_chunk(id, session, offset, content_length, body) + .await?) + } + + /// Auth-aware wrapper around [`StorageService::upload_offset`]. + pub async fn upload_offset( + &self, + id: ObjectId, + session: SessionToken, + ) -> ApiResult { + self.check_permission(Permission::ObjectWrite, id.context())?; + Ok(self.service.upload_offset(id, session).await?) + } + + /// Auth-aware wrapper around [`StorageService::terminate_upload`]. + pub async fn terminate_upload( + &self, + id: ObjectId, + session: SessionToken, + ) -> ApiResult { + self.check_permission(Permission::ObjectWrite, id.context())?; + Ok(self.service.terminate_upload(id, session).await?) + } } diff --git a/objectstore-server/src/endpoints/common.rs b/objectstore-server/src/endpoints/common.rs index 97468764..75b89cbe 100644 --- a/objectstore-server/src/endpoints/common.rs +++ b/objectstore-server/src/endpoints/common.rs @@ -52,6 +52,17 @@ pub struct ApiErrorResponse { } impl ApiErrorResponse { + /// Creates an error response carrying only a message, with no cause chain. + /// + /// For outcomes that are not errors in the service layer and therefore have no + /// [`Error`] to wrap, such as a denied resumable upload session. + pub fn message(detail: impl Into) -> Self { + Self { + detail: Some(detail.into()), + causes: Vec::new(), + } + } + /// Creates an error response from an error, extracting the full cause chain. pub fn from_error(error: &E) -> Self { let detail = Some(error.to_string()); @@ -96,6 +107,9 @@ impl ApiError { StatusCode::RANGE_NOT_SATISFIABLE } ApiError::Service(ServiceError::InvalidUploadId(_)) => StatusCode::BAD_REQUEST, + ApiError::Service(ServiceError::InvalidUploadRequest(_)) => StatusCode::BAD_REQUEST, + ApiError::Service(ServiceError::UploadOffsetMismatch { .. }) => StatusCode::CONFLICT, + ApiError::Service(ServiceError::UploadSessionGone) => StatusCode::GONE, ApiError::Service(ServiceError::AtCapacity) => StatusCode::TOO_MANY_REQUESTS, ApiError::Service(ServiceError::NotImplemented) => StatusCode::NOT_IMPLEMENTED, ApiError::Service(_) => StatusCode::INTERNAL_SERVER_ERROR, diff --git a/objectstore-server/src/endpoints/mod.rs b/objectstore-server/src/endpoints/mod.rs index ab460b69..da96257a 100644 --- a/objectstore-server/src/endpoints/mod.rs +++ b/objectstore-server/src/endpoints/mod.rs @@ -1,5 +1,113 @@ //! Contains all HTTP endpoint handlers. //! +//! This module documents the request and response shape of every route; see the [crate +//! documentation](crate) for the layers a request passes through before reaching a handler. +//! +//! Scopes are encoded in the URL path using Matrix URI syntax: `org=123;project=456`. An +//! underscore (`_`) represents empty scopes. +//! +//! # Object Endpoints +//! +//! All object operations live under the `/v1/` prefix: +//! +//! | Method | Path | Description | +//! |----------|-------------------------------------------|------------------------------| +//! | `POST` | `/v1/objects/{usecase}/{scopes}/` | Insert with server-generated key | +//! | `GET` | `/v1/objects/{usecase}/{scopes}/{*key}` | Retrieve object | +//! | `HEAD` | `/v1/objects/{usecase}/{scopes}/{*key}` | Retrieve metadata only | +//! | `PUT` | `/v1/objects/{usecase}/{scopes}/{*key}` | Insert or overwrite with key | +//! | `DELETE` | `/v1/objects/{usecase}/{scopes}/{*key}` | Delete object | +//! | `POST` | `/v1/objects:batch/{usecase}/{scopes}/` | Batch operations (multipart) | +//! +//! Object metadata travels in request and response headers; see +//! [`objectstore_types::metadata`] for the mapping. +//! +//! # Resumable Upload Endpoints +//! +//! A resumable upload transfers a single object across several requests. The client opens a +//! session, declaring the object's total size and metadata upfront, and then sends the payload +//! as a sequence of chunks at increasing byte offsets. If a chunk fails, the client asks the +//! server which offset it holds and continues from there, so an interrupted transfer resumes +//! where it stopped instead of starting over. The server knows the total size from the +//! session, so it recognizes the chunk carrying the last byte and commits the object itself. +//! +//! Resumable uploads use the object endpoints above, selected by a query parameter: +//! `upload_type=resumable` opens a session, and `session=` addresses it from then on. +//! The object is named by the request path as usual, and [`objectstore_types::resumable`] +//! holds the protocol types. +//! +//! | Method | Path | Description | +//! |----------|------------------------------------------------------------|----------------------------------------------| +//! | `POST` | `/v1/objects/{usecase}/{scopes}/?upload_type=resumable` | Create session (server-generated key) | +//! | `PUT` | `/v1/objects/{usecase}/{scopes}/{*key}?upload_type=resumable` | Create session (user-provided key) | +//! | `PUT` | `/v1/objects/{usecase}/{scopes}/{*key}?session=` | Upload a chunk, or query the offset | +//! | `DELETE` | `/v1/objects/{usecase}/{scopes}/{*key}?session=` | Terminate session, discarding what was sent | +//! +//! Session creation requires an `Upload-Length` header carrying the total size of the object +//! in bytes, and takes the same metadata headers as a regular upload. It answers `200 OK` +//! with `{"key", "session"}` and a `Location` header pointing at the object path with the +//! session appended. Metadata is fixed at this point and does not change afterwards. +//! +//! Chunk uploads and offset queries share one request shape, distinguished by the +//! `Upload-Offset` header: a byte offset submits the body as the chunk starting there, while +//! the `*` wildcard submits an empty body and asks which offset the server holds. Both answer +//! `204 No Content` with the authoritative `Upload-Offset` while bytes remain, and +//! `201 Created` with `{"key"}` once the object is committed. **The offset in the response +//! may be lower than the end of the chunk that was sent** — backends persist only aligned +//! prefixes and discard the remainder — so clients always continue from the returned offset. +//! +//! An offset query can commit an object that was assembled but not yet committed, so it +//! requires write permission despite being read-shaped. Termination likewise needs write +//! rather than delete permission: it releases an in-progress upload, not an object. +//! +//! | Status | Meaning | Client action | +//! |--------|---------|---------------| +//! | `400` | Malformed: unusable session, missing `Upload-Length`, or a chunk exceeding the declared length | Terminal | +//! | `409` | On creation: resumable uploads are unavailable for this object. On a chunk: offset mismatch, with the authoritative offset in `Upload-Offset` | Fall back to a regular upload, or resynchronize | +//! | `410` | The session expired or was terminated; nothing was retained | Start a new session | +//! | `501` | The configured backend does not implement resumable uploads | Fall back to a regular upload | +//! +//! Not every backend can support this. Session creation asks the backend that would store the +//! object to open one, and a backend that cannot declines, which the server reports as +//! `409 Conflict`. No backend implements resumable uploads yet, so every session creation is +//! currently denied. +//! +//! # Multipart Upload Endpoints +//! +//! Multipart uploads are being replaced by [resumable +//! uploads](#resumable-upload-endpoints) and will be removed once all consumers have +//! migrated. See [`objectstore_types::multipart`] for the protocol types. +//! +//! | Method | Path | Description | +//! |-----------|--------------------------------------------------------------|--------------------------------------| +//! | `POST` | `/v1/objects:multipart/{usecase}/{scopes}/` | Initiate upload (server-generated key) | +//! | `PUT` | `/v1/objects:multipart/{usecase}/{scopes}/{*key}` | Initiate upload (user-provided key) | +//! | `PUT` | `/v1/objects:multipart:parts/{usecase}/{scopes}/{*key}` | Upload a part (`upload_id`, `part_number` query params) | +//! | `GET` | `/v1/objects:multipart:parts/{usecase}/{scopes}/{*key}` | List uploaded parts (`upload_id` query param) | +//! | `POST` | `/v1/objects:multipart:complete/{usecase}/{scopes}/{*key}` | Complete upload (`upload_id` query param) | +//! | `DELETE` | `/v1/objects:multipart/{usecase}/{scopes}/{*key}` | Abort upload (`upload_id` query param) | +//! +//! The initiate POST endpoint accepts both trailing-slash and non-trailing-slash forms. +//! +//! The complete endpoint returns `200 OK` immediately, with a streaming body that will +//! contain the error (if any) as JSON. Whitespace is sent in the streaming body to keep the +//! connection open. Clients must parse the body to determine the actual outcome, and not rely +//! on the status code. +//! +//! # Internal Endpoints +//! +//! Internal endpoints are exempt from authentication, rate limiting, and the web concurrency +//! limit so they remain available when the server is under load. [`is_internal_route`] +//! identifies them. +//! +//! | Method | Path | Description | +//! |--------|------|-------------| +//! | `GET` | `/health` | Liveness probe (always returns 200) | +//! | `GET` | `/ready` | Readiness probe (returns 503 when `/tmp/objectstore.down` exists, enabling graceful drain) | +//! | `GET` | `/keda` | Prometheus text-format gauges for KEDA autoscaling (see [KEDA Metrics](crate#keda-metrics)) | +//! +//! # Code Usage +//! //! Use [`routes`] to create a router with all endpoints. use axum::Router; @@ -14,6 +122,7 @@ mod multipart; mod objects; #[cfg(all(target_os = "linux", feature = "profiling"))] mod profiling; +mod resumable; /// Returns `true` for internal endpoints that are exempt from metrics and concurrency limits. pub fn is_internal_route(route: &str) -> bool { diff --git a/objectstore-server/src/endpoints/objects.rs b/objectstore-server/src/endpoints/objects.rs index 1aafcbac..7ee6728e 100644 --- a/objectstore-server/src/endpoints/objects.rs +++ b/objectstore-server/src/endpoints/objects.rs @@ -1,7 +1,7 @@ use std::fmt::Write as _; use axum::body::Body; -use axum::extract::State; +use axum::extract::{OriginalUri, Query, State}; use axum::http::{HeaderMap, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::routing; @@ -15,6 +15,7 @@ use serde::Serialize; use crate::auth::AuthAwareService; use crate::endpoints::common::{ApiError, ApiResult, insert_accept_ranges}; +use crate::endpoints::resumable::{self, RequestPath, ResumableQuery, ResumableRoute}; use crate::extractors::byte_range::OptionalByteRange; use crate::extractors::{Xt, body::MeteredBody}; use crate::state::ServiceState; @@ -43,9 +44,27 @@ async fn objects_post( service: AuthAwareService, State(state): State, Xt(context): Xt, + OriginalUri(uri): OriginalUri, + Query(query): Query, headers: HeaderMap, MeteredBody(body): MeteredBody, ) -> ApiResult { + // A chunk always addresses a resolved key, so `?session=` has no meaning on the + // collection route. `?upload_type=resumable` creates a session for a generated key. + match query.classify()? { + ResumableRoute::Create => { + let id = ObjectId::optional(context, None); + let path = RequestPath::Collection; + return resumable::create_session(service, state, path, uri.path(), id, headers).await; + } + ResumableRoute::Session(_) => { + return Err(ApiError::Client( + "`session` requires an object key; use PUT on the object path".into(), + )); + } + ResumableRoute::Regular => {} + } + let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?; state @@ -199,9 +218,26 @@ async fn object_put( service: AuthAwareService, State(state): State, Xt(id): Xt, + OriginalUri(uri): OriginalUri, + Query(query): Query, headers: HeaderMap, - MeteredBody(body): MeteredBody, + body: MeteredBody, ) -> ApiResult { + // `PUT` carries all three write shapes: create a session, write a chunk, query the + // offset. `MeteredBody` is extracted unconditionally and dropped unread on the two + // bodyless paths. + match query.classify()? { + ResumableRoute::Create => { + let path = RequestPath::Object; + return resumable::create_session(service, state, path, uri.path(), id, headers).await; + } + ResumableRoute::Session(session) => { + return resumable::session_request(service, id, session, headers, body).await; + } + ResumableRoute::Regular => {} + } + + let MeteredBody(body) = body; let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?; let ObjectId { context, key } = id; @@ -226,7 +262,14 @@ async fn object_put( async fn object_delete( service: AuthAwareService, Xt(id): Xt, -) -> ApiResult { + Query(query): Query, +) -> ApiResult { + // With a session this terminates the upload; without one it deletes the object, as it + // always has. Note the two need different permissions — see `AuthAwareService`. + if let ResumableRoute::Session(session) = query.classify_session_only("DELETE")? { + return resumable::terminate(service, id, session).await; + } + service.delete_object(id).await?; - Ok(StatusCode::NO_CONTENT) + Ok(StatusCode::NO_CONTENT.into_response()) } diff --git a/objectstore-server/src/endpoints/resumable.rs b/objectstore-server/src/endpoints/resumable.rs new file mode 100644 index 00000000..897e0207 --- /dev/null +++ b/objectstore-server/src/endpoints/resumable.rs @@ -0,0 +1,491 @@ +//! Resumable upload endpoints. +//! +//! Resumable uploads are a variation of the regular object endpoints rather than a separate +//! resource, following GCS and S3 rather than [TUS]. Every request addresses the same object +//! path with the session in the query string, so these handlers have no router of their own: +//! [`objects`](super::objects) dispatches to them based on [`ResumableQuery`]. +//! +//! | Operation | Request | Success | +//! |---|---|---| +//! | Create | `POST /objects/{usecase}/{scopes}/?upload_type=resumable` | `200` + `Location` + `{"key","session"}` | +//! | Create | `PUT /objects/{usecase}/{scopes}/{key}?upload_type=resumable` | `200` + `Location` + `{"key","session"}` | +//! | Chunk | `PUT …/{key}?session=` with `Upload-Offset: ` | `204` + `Upload-Offset`, or `201` + `{"key"}` | +//! | Offset query | `PUT …/{key}?session=` with `Upload-Offset: *` | `204` + `Upload-Offset`, or `201` + `{"key"}` | +//! | Terminate | `DELETE …/{key}?session=` | `204` | +//! +//! There is no completion request. The total size is known from session creation, so the +//! backend recognizes the chunk carrying the last byte and commits the object itself. +//! +//! Not every backend supports this. When one declines, session creation answers +//! `409 Conflict` and the client performs a regular upload instead. +//! +//! [TUS]: https://tus.io/protocols/resumable-upload + +use axum::http::{HeaderMap, StatusCode}; +use axum::response::{IntoResponse, Response}; +use axum::{Json, http}; +use objectstore_service::error::Error as ServiceError; +use objectstore_service::id::ObjectId; +use objectstore_service::resumable::{SessionToken, UploadOffset, UploadProgress}; +use objectstore_types::metadata::Metadata; +use objectstore_types::resumable::{ + CommitResponse, CreateSessionResponse, HEADER_UPLOAD_LENGTH, HEADER_UPLOAD_OFFSET, +}; +use serde::Deserialize; + +use crate::auth::AuthAwareService; +use crate::endpoints::common::{ApiError, ApiErrorResponse, ApiResult}; +use crate::extractors::body::MeteredBody; +use crate::state::ServiceState; + +/// The `upload_type` query parameter. +/// +/// Only one value is accepted, so an unrecognized upload type is a deserialization failure +/// and therefore a `400` rather than being silently treated as a regular upload. +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub(super) enum UploadType { + /// Create a resumable upload session. + Resumable, +} + +/// The resumable protocol's query parameters, as seen on a regular object route. +/// +/// Both fields are optional and unknown parameters are ignored, because pre-signed URLs put +/// their own `os_*` parameters into the same query string. +#[derive(Debug, Default, Deserialize)] +pub(super) struct ResumableQuery { + /// Present on a session creation request. + upload_type: Option, + /// Present on a chunk write, offset query, or termination. + session: Option, +} + +/// What a request on an object route is addressing. +#[derive(Debug)] +pub(super) enum ResumableRoute { + /// Create a session for the object named by the request path. + Create, + /// Act on the identified session: write a chunk, query the offset, or terminate. + Session(SessionToken), + /// A regular object request that does not involve the resumable protocol. + Regular, +} + +impl ResumableQuery { + /// Classifies a request that may create a session or act on one. + /// + /// # Errors + /// + /// Returns [`ApiError::Client`] if both parameters are present. They address different + /// operations, so a request carrying both is ambiguous rather than defaulted. + pub fn classify(self) -> ApiResult { + match (self.upload_type, self.session) { + (Some(_), Some(_)) => Err(ApiError::Client( + "`upload_type` and `session` are mutually exclusive".into(), + )), + (Some(UploadType::Resumable), None) => Ok(ResumableRoute::Create), + (None, Some(session)) => Ok(ResumableRoute::Session(session)), + (None, None) => Ok(ResumableRoute::Regular), + } + } + + /// Classifies a request that may only act on an existing session. + /// + /// Used by routes where session creation is not defined: `DELETE`, which terminates, and + /// the collection `POST`, whose generated key is only known once a session exists. + /// + /// # Errors + /// + /// Returns [`ApiError::Client`] if `upload_type` is present. + pub fn classify_session_only(self, operation: &str) -> ApiResult { + if self.upload_type.is_some() { + return Err(ApiError::Client(format!( + "`upload_type` is not supported on {operation}" + ))); + } + + match self.session { + Some(session) => Ok(ResumableRoute::Session(session)), + None => Ok(ResumableRoute::Regular), + } + } +} + +/// Reads the required [`HEADER_UPLOAD_LENGTH`] header. +fn upload_length(headers: &HeaderMap) -> ApiResult { + let value = headers + .get(HEADER_UPLOAD_LENGTH) + .ok_or_else(|| ApiError::Client(format!("{HEADER_UPLOAD_LENGTH} header is required")))?; + + value + .to_str() + .ok() + .filter(|v| v.bytes().all(|b| b.is_ascii_digit())) + .and_then(|v| v.parse().ok()) + .ok_or_else(|| ApiError::Client(format!("{HEADER_UPLOAD_LENGTH} must be a byte count"))) +} + +/// Reads the required [`HEADER_UPLOAD_OFFSET`] header. +fn upload_offset(headers: &HeaderMap) -> ApiResult { + let value = headers + .get(HEADER_UPLOAD_OFFSET) + .ok_or_else(|| ApiError::Client(format!("{HEADER_UPLOAD_OFFSET} header is required")))?; + + value + .to_str() + .map_err(|_| ApiError::Client(format!("{HEADER_UPLOAD_OFFSET} must be ASCII")))? + .parse() + .map_err(|e: objectstore_types::resumable::InvalidUploadOffset| { + ApiError::Client(e.to_string()) + }) +} + +/// Reads the required `Content-Length` header. +/// +/// Chunks declare their length so the server can forward only the prefix a backend accepts +/// without buffering the body to find out how long it is. +fn content_length(headers: &HeaderMap) -> ApiResult { + headers + .get(http::header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .ok_or_else(|| ApiError::Client("Content-Length header is required".into())) +} + +/// How the request path relates to the object a session is being created for. +/// +/// Needed to build the `Location` header, which must always name the object even when the +/// request did not. +#[derive(Clone, Copy, Debug)] +pub(super) enum RequestPath { + /// The request path names the object, as on a `PUT` to the object route. + Object, + /// The request path is the collection the object lives in, as on a `POST` whose key was + /// generated by the server and therefore does not appear in the path. + Collection, +} + +/// Creates a session for the object at `id`. +/// +/// Answers `409 Conflict` when the backend declines, which tells the client to fall back to a +/// regular upload. Metadata is declared here and does not change afterwards. +pub(super) async fn create_session( + service: AuthAwareService, + state: ServiceState, + request_path: RequestPath, + uri_path: &str, + id: ObjectId, + headers: HeaderMap, +) -> ApiResult { + let total_length = upload_length(&headers)?; + let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?; + + state + .config + .usecases + .validate(&id.context().usecase, &metadata) + .map_err(|e| ApiError::Client(e.to_string()))?; + + let Some(session) = service + .create_upload_session(id.clone(), metadata, total_length) + .await? + else { + let body = ApiErrorResponse::message("resumable uploads are unavailable for this object"); + return Ok((StatusCode::CONFLICT, Json(body)).into_response()); + }; + + let mut headers = HeaderMap::new(); + if let Some(location) = session_location(uri_path, request_path, id.key(), &session) { + headers.insert(http::header::LOCATION, location); + } + + let body = Json(CreateSessionResponse { + key: id.key().to_owned(), + session, + }); + Ok((StatusCode::OK, headers, body).into_response()) +} + +/// Builds the `Location` header pointing at the session. +/// +/// The value is the object path with the session appended, so a client that persisted the key +/// and the session can rebuild it without having stored a URL. `Router::nest` strips the `/v1` +/// prefix from the request URI, so callers pass the path from +/// [`OriginalUri`](axum::extract::OriginalUri) — and on the collection route that path does +/// not name the object, so the key is appended to it. +/// +/// Returns `None` if the result is not a valid header value, in which case the header is +/// omitted: it is a convenience, and the response body carries the same information. +fn session_location( + uri_path: &str, + request_path: RequestPath, + key: &str, + session: &SessionToken, +) -> Option { + let object_path = match request_path { + RequestPath::Object => uri_path.to_owned(), + RequestPath::Collection => format!("{}/{key}", uri_path.trim_end_matches('/')), + }; + + http::HeaderValue::from_str(&format!("{object_path}?session={session}")).ok() +} + +/// Acts on an open session: writes a chunk, or reports the offset the server holds. +/// +/// [`HEADER_UPLOAD_OFFSET`] selects between the two. A concrete offset submits the request +/// body as the chunk starting there; the `*` wildcard submits nothing and asks where the +/// server stands, which also commits an object that was assembled but not yet committed. +/// +/// Both answer `204 No Content` with the authoritative offset while bytes remain, and +/// `201 Created` with the key once the object is committed. +pub(super) async fn session_request( + service: AuthAwareService, + id: ObjectId, + session: SessionToken, + headers: HeaderMap, + MeteredBody(body): MeteredBody, +) -> ApiResult { + let offset = upload_offset(&headers)?; + let content_length = content_length(&headers)?; + let key = id.key().to_owned(); + + let progress = match offset { + UploadOffset::At(offset) => { + service + .put_chunk(id, session, offset, content_length, body) + .await + } + UploadOffset::Unknown => { + // The wildcard carries no payload. A body would be silently discarded, so + // reject it rather than let a client believe those bytes were written. + if content_length != 0 { + return Err(ApiError::Client(format!( + "{HEADER_UPLOAD_OFFSET}: * must be sent with an empty body" + ))); + } + + service.upload_offset(id, session).await + } + }; + + progress_response(progress, key) +} + +/// Terminates a session, discarding whatever was uploaded. +pub(super) async fn terminate( + service: AuthAwareService, + id: ObjectId, + session: SessionToken, +) -> ApiResult { + service.terminate_upload(id, session).await?; + Ok(StatusCode::NO_CONTENT.into_response()) +} + +/// Turns an [`UploadProgress`] outcome into the response shared by chunks and offset queries. +/// +/// An offset mismatch is answered here rather than through [`ApiError::status`], because the +/// authoritative offset has to travel in a header that a generic error response cannot set. +fn progress_response(progress: ApiResult, key: String) -> ApiResult { + let progress = match progress { + Ok(progress) => progress, + Err(ApiError::Service(ServiceError::UploadOffsetMismatch { offset })) => { + let body = ApiErrorResponse::message(format!("expected offset {offset}")); + let response = ( + StatusCode::CONFLICT, + [(HEADER_UPLOAD_OFFSET, http::HeaderValue::from(offset))], + Json(body), + ); + return Ok(response.into_response()); + } + Err(e) => return Err(e), + }; + + let response = match progress { + UploadProgress::Incomplete { offset } => ( + StatusCode::NO_CONTENT, + [(HEADER_UPLOAD_OFFSET, http::HeaderValue::from(offset))], + ) + .into_response(), + UploadProgress::Committed => { + (StatusCode::CREATED, Json(CommitResponse { key })).into_response() + } + }; + + Ok(response) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn query(upload_type: Option, session: Option<&str>) -> ResumableQuery { + ResumableQuery { + upload_type, + session: session.map(|s| SessionToken::new(s.into()).unwrap()), + } + } + + #[test] + fn classify_recognizes_each_operation() { + assert!(matches!( + query(Some(UploadType::Resumable), None).classify(), + Ok(ResumableRoute::Create) + )); + assert!(matches!( + query(None, Some("token")).classify(), + Ok(ResumableRoute::Session(_)) + )); + assert!(matches!( + query(None, None).classify(), + Ok(ResumableRoute::Regular) + )); + } + + #[test] + fn classify_rejects_both_parameters() { + let result = query(Some(UploadType::Resumable), Some("token")).classify(); + assert!(matches!(result, Err(ApiError::Client(_))), "{result:?}"); + } + + #[test] + fn classify_session_only_rejects_upload_type() { + let result = query(Some(UploadType::Resumable), None).classify_session_only("DELETE"); + assert!(matches!(result, Err(ApiError::Client(_))), "{result:?}"); + + assert!(matches!( + query(None, Some("token")).classify_session_only("DELETE"), + Ok(ResumableRoute::Session(_)) + )); + assert!(matches!( + query(None, None).classify_session_only("DELETE"), + Ok(ResumableRoute::Regular) + )); + } + + #[test] + fn upload_length_requires_a_byte_count() { + let mut headers = HeaderMap::new(); + assert!(upload_length(&headers).is_err(), "missing header"); + + for invalid in ["", "-1", "+1", "1.5", "abc", " 1"] { + headers.insert(HEADER_UPLOAD_LENGTH, invalid.parse().unwrap()); + assert!(upload_length(&headers).is_err(), "accepted {invalid:?}"); + } + + headers.insert(HEADER_UPLOAD_LENGTH, "1048576".parse().unwrap()); + assert_eq!(upload_length(&headers).unwrap(), 1_048_576); + } + + #[test] + fn upload_offset_parses_chunk_and_wildcard() { + let mut headers = HeaderMap::new(); + assert!(upload_offset(&headers).is_err(), "missing header"); + + headers.insert(HEADER_UPLOAD_OFFSET, "*".parse().unwrap()); + assert_eq!(upload_offset(&headers).unwrap(), UploadOffset::Unknown); + + headers.insert(HEADER_UPLOAD_OFFSET, "262144".parse().unwrap()); + assert_eq!(upload_offset(&headers).unwrap(), UploadOffset::At(262_144)); + + headers.insert(HEADER_UPLOAD_OFFSET, "nope".parse().unwrap()); + assert!(upload_offset(&headers).is_err()); + } + + /// Reads a response's status, `Upload-Offset` header, and body. + async fn parts_of(response: Response) -> (StatusCode, Option, String) { + let status = response.status(); + let offset = response + .headers() + .get(HEADER_UPLOAD_OFFSET) + .map(|v| v.to_str().unwrap().to_owned()); + + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + + (status, offset, String::from_utf8(body.to_vec()).unwrap()) + } + + #[tokio::test] + async fn incomplete_progress_answers_no_content_with_the_offset() { + let progress = Ok(UploadProgress::Incomplete { offset: 262_144 }); + let response = progress_response(progress, "my-key".into()).unwrap(); + + let (status, offset, body) = parts_of(response).await; + assert_eq!(status, StatusCode::NO_CONTENT); + assert_eq!(offset.as_deref(), Some("262144")); + assert!(body.is_empty(), "204 must not carry a body: {body:?}"); + } + + #[tokio::test] + async fn commit_answers_created_with_the_key() { + let response = progress_response(Ok(UploadProgress::Committed), "my-key".into()).unwrap(); + + let (status, offset, body) = parts_of(response).await; + assert_eq!(status, StatusCode::CREATED); + assert_eq!(offset, None, "a commit reports no offset"); + assert_eq!(body, r#"{"key":"my-key"}"#); + } + + #[tokio::test] + async fn offset_mismatch_answers_conflict_with_the_authoritative_offset() { + let mismatch = ServiceError::UploadOffsetMismatch { offset: 786_432 }; + let response = + progress_response(Err(ApiError::Service(mismatch)), "my-key".into()).unwrap(); + + let (status, offset, body) = parts_of(response).await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!( + offset.as_deref(), + Some("786432"), + "the client resynchronizes from this header" + ); + assert!(body.contains("786432"), "{body:?}"); + } + + #[tokio::test] + async fn other_errors_propagate_unchanged() { + let gone = ApiError::Service(ServiceError::UploadSessionGone); + let error = progress_response(Err(gone), "my-key".into()).unwrap_err(); + assert_eq!(error.status(), StatusCode::GONE); + + let invalid = ApiError::Service(ServiceError::InvalidUploadRequest("bad".into())); + let error = progress_response(Err(invalid), "my-key".into()).unwrap_err(); + assert_eq!(error.status(), StatusCode::BAD_REQUEST); + } + + #[test] + fn session_location_appends_the_session_to_the_object_path() { + let session = SessionToken::new("tok3n".into()).unwrap(); + let location = session_location( + "/v1/objects/testing/org=1/my-key", + RequestPath::Object, + "my-key", + &session, + ); + + assert_eq!( + location.unwrap(), + "/v1/objects/testing/org=1/my-key?session=tok3n" + ); + } + + #[test] + fn session_location_appends_a_generated_key_to_the_collection_path() { + let session = SessionToken::new("tok3n".into()).unwrap(); + + // The `POST` route matches with and without a trailing slash, and the generated key + // never appears in the request path — so it has to be appended either way. + for collection in ["/v1/objects/testing/org=1/", "/v1/objects/testing/org=1"] { + let location = + session_location(collection, RequestPath::Collection, "generated", &session); + + assert_eq!( + location.unwrap(), + "/v1/objects/testing/org=1/generated?session=tok3n", + "for request path {collection:?}" + ); + } + } +} diff --git a/objectstore-server/tests/resumable.rs b/objectstore-server/tests/resumable.rs new file mode 100644 index 00000000..72d6a4bc --- /dev/null +++ b/objectstore-server/tests/resumable.rs @@ -0,0 +1,312 @@ +//! Integration tests for the resumable upload endpoints. +//! +//! No backend implements resumable uploads yet, so the reachable surface is session denial +//! and request validation. That is deliberate: a deployment must answer `409 Conflict` to +//! every session creation so clients fall back to a regular upload, and it must reject a +//! malformed request before it reaches a backend. +//! +//! The `501 Not Implemented` assertions are the proof that dispatch and header parsing work: +//! the only way to reach a declining backend method is through a well-formed request. + +use anyhow::Result; +use objectstore_server::config::{AuthZ, Config}; +use objectstore_test::server::TestServer; +use objectstore_types::resumable::{HEADER_UPLOAD_LENGTH, HEADER_UPLOAD_OFFSET}; +use reqwest::StatusCode; + +async fn test_server() -> TestServer { + TestServer::with_config(Config { + auth: AuthZ { + enforce: false, + ..Default::default() + }, + ..Default::default() + }) + .await +} + +// --- Session creation --- + +#[tokio::test] +async fn create_session_is_denied_with_client_key() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?upload_type=resumable")) + .header(HEADER_UPLOAD_LENGTH, "1048576") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::CONFLICT); + Ok(()) +} + +#[tokio::test] +async fn create_session_is_denied_with_generated_key() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .post(server.url("/v1/objects/test/org=1/?upload_type=resumable")) + .header(HEADER_UPLOAD_LENGTH, "1048576") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::CONFLICT); + Ok(()) +} + +#[tokio::test] +async fn create_session_requires_upload_length() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?upload_type=resumable")) + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +#[tokio::test] +async fn create_session_rejects_malformed_upload_length() -> Result<()> { + let server = test_server().await; + let client = reqwest::Client::new(); + + for invalid in ["", "-1", "1.5", "lots"] { + let response = client + .put(server.url("/v1/objects/test/org=1/my-key?upload_type=resumable")) + .header(HEADER_UPLOAD_LENGTH, invalid) + .send() + .await?; + + assert_eq!( + response.status(), + StatusCode::BAD_REQUEST, + "accepted {HEADER_UPLOAD_LENGTH}: {invalid:?}" + ); + } + + Ok(()) +} + +#[tokio::test] +async fn unknown_upload_type_is_rejected() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?upload_type=multipart")) + .header(HEADER_UPLOAD_LENGTH, "1048576") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +// --- Chunks and offset queries --- + +#[tokio::test] +async fn chunk_reaches_the_declining_backend() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?session=some-token")) + .header(HEADER_UPLOAD_OFFSET, "0") + .body("payload") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::NOT_IMPLEMENTED); + Ok(()) +} + +#[tokio::test] +async fn offset_query_reaches_the_declining_backend() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?session=some-token")) + .header(HEADER_UPLOAD_OFFSET, "*") + .header(reqwest::header::CONTENT_LENGTH, "0") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::NOT_IMPLEMENTED); + Ok(()) +} + +#[tokio::test] +async fn chunk_requires_upload_offset() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?session=some-token")) + .body("payload") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +#[tokio::test] +async fn chunk_rejects_malformed_upload_offset() -> Result<()> { + let server = test_server().await; + let client = reqwest::Client::new(); + + for invalid in ["", "-1", "1.5", "**", "here"] { + let response = client + .put(server.url("/v1/objects/test/org=1/my-key?session=some-token")) + .header(HEADER_UPLOAD_OFFSET, invalid) + .body("payload") + .send() + .await?; + + assert_eq!( + response.status(), + StatusCode::BAD_REQUEST, + "accepted {HEADER_UPLOAD_OFFSET}: {invalid:?}" + ); + } + + Ok(()) +} + +#[tokio::test] +async fn offset_query_rejects_a_payload() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?session=some-token")) + .header(HEADER_UPLOAD_OFFSET, "*") + .body("payload") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +#[tokio::test] +async fn session_token_with_path_traversal_is_rejected() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?session=../escape")) + .header(HEADER_UPLOAD_OFFSET, "0") + .body("payload") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +// --- Termination --- + +#[tokio::test] +async fn terminate_reaches_the_declining_backend() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .delete(server.url("/v1/objects/test/org=1/my-key?session=some-token")) + .send() + .await?; + + assert_eq!(response.status(), StatusCode::NOT_IMPLEMENTED); + Ok(()) +} + +#[tokio::test] +async fn delete_rejects_upload_type() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .delete(server.url("/v1/objects/test/org=1/my-key?upload_type=resumable")) + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +// --- Parameter combinations --- + +#[tokio::test] +async fn upload_type_and_session_are_mutually_exclusive() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key?upload_type=resumable&session=some-token")) + .header(HEADER_UPLOAD_LENGTH, "1048576") + .header(HEADER_UPLOAD_OFFSET, "0") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +#[tokio::test] +async fn session_on_the_collection_route_is_rejected() -> Result<()> { + let server = test_server().await; + + let response = reqwest::Client::new() + .post(server.url("/v1/objects/test/org=1/?session=some-token")) + .header(HEADER_UPLOAD_OFFSET, "0") + .body("payload") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +// --- Regular uploads are unaffected --- + +#[tokio::test] +async fn regular_object_operations_still_work() -> Result<()> { + let server = test_server().await; + let client = reqwest::Client::new(); + + let response = client + .put(server.url("/v1/objects/test/org=1/my-key")) + .body("payload") + .send() + .await?; + assert_eq!(response.status(), StatusCode::OK); + + let response = client + .get(server.url("/v1/objects/test/org=1/my-key")) + .send() + .await?; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.text().await?, "payload"); + + let response = client + .delete(server.url("/v1/objects/test/org=1/my-key")) + .send() + .await?; + assert_eq!(response.status(), StatusCode::NO_CONTENT); + + Ok(()) +} + +#[tokio::test] +async fn regular_upload_ignores_resumable_headers() -> Result<()> { + let server = test_server().await; + + // Without a query parameter the request is a regular upload, and the protocol headers + // carry no meaning. They must not accidentally engage the resumable path. + let response = reqwest::Client::new() + .put(server.url("/v1/objects/test/org=1/my-key")) + .header(HEADER_UPLOAD_LENGTH, "7") + .header(HEADER_UPLOAD_OFFSET, "0") + .body("payload") + .send() + .await?; + + assert_eq!(response.status(), StatusCode::OK); + Ok(()) +} diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 0b745d46..6f98a241 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -190,6 +190,30 @@ The default execution limit is [`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). See [`StorageService::with_concurrency`] for configuration. +## Resumable Uploads + +A resumable upload writes one object across several requests: the payload arrives +as a sequence of chunks at increasing byte offsets, so an interrupted transfer +continues where it stopped instead of starting over. That is worth the extra round +trips for objects large enough that re-sending the whole payload is expensive. + +[`StorageService`] exposes four operations, each a method on +[`Backend`](backend::common::Backend), which run in this sequence: + +1. [`create_upload_session`](backend::common::Backend::create_upload_session) + declares the total size and metadata, and returns a session token. +2. [`put_chunk`](backend::common::Backend::put_chunk) writes bytes at an offset and + reports the offset now persisted. +3. After a failure, [`upload_offset`](backend::common::Backend::upload_offset) + reports where the backend stands, so the caller resumes from there. +4. The chunk carrying the last byte commits the object. There is no finalize call — + the backend recognizes that chunk from the declared total size. +5. At any time, a session can be terminated, which discards what it holds. + +Not all backends support resumable uploads and can decline creating a session. +Support can depend on the declared size, the metadata, or whether resuming is +possible in principle. + ## Multipart Uploads When the configured backend supports it, [`StorageService`] exposes multipart diff --git a/objectstore-service/src/backend/common.rs b/objectstore-service/src/backend/common.rs index 45b9b4a0..e29e27dd 100644 --- a/objectstore-service/src/backend/common.rs +++ b/objectstore-service/src/backend/common.rs @@ -13,6 +13,9 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::{ + CreateSessionResponse, SessionToken, TerminateUploadResponse, UploadProgress, +}; use crate::stream::{ClientStream, PayloadStream}; /// User agent string used for outgoing requests. @@ -72,6 +75,112 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static { fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> { Err(Error::NotImplemented) } + + /// Opens a resumable upload session for the object at `id`. + /// + /// `total_length` is the complete size of the object in bytes, declared by the client + /// when the session is created. It is a parameter of its own rather than part of + /// `metadata`, because [`Metadata::size`] is materialized by the server and never + /// trusted from a client. The backend needs it to recognize the final chunk, and a + /// tiering backend needs it to decide where the object would be placed. + /// + /// `metadata` is fixed for the lifetime of the session and does not change afterwards. + /// Compression is recorded rather than applied: the payload must already be compressed, + /// since its total length has to be known at this point. + /// + /// Returns `Ok(None)` when this backend cannot store the described object resumably. + /// Declining is a routine outcome, not an error — the server denies the session and the + /// client falls back to a regular upload. The default implementation declines, so a + /// backend opts in simply by overriding this method. There is deliberately no separate + /// capability trait and no probe: support can depend on the size, the metadata and the + /// routing result at once, all of which are only known here. + /// + /// # Errors + /// + /// Returns an error only when the backend supports resumable uploads but failed to open + /// the session. + async fn create_upload_session( + &self, + id: &ObjectId, + metadata: &Metadata, + total_length: u64, + ) -> Result { + let _ = (id, metadata, total_length); + Ok(None) + } + + /// Writes a chunk of `content_length` bytes at `offset` into an open session. + /// + /// `offset` must equal the offset the backend currently holds. Backends persist only + /// aligned prefixes and discard the remainder, so the offset in the returned + /// [`UploadProgress::Incomplete`] is authoritative and may be lower than + /// `offset + content_length`. + /// + /// A session has a single writer. Concurrent chunk writes are not coordinated: one of + /// them wins and the others fail with [`Error::UploadOffsetMismatch`]. + /// + /// Once the chunk carrying the last byte is persisted, the backend assembles and commits + /// the object and returns [`UploadProgress::Committed`]. + /// + /// # Errors + /// + /// - [`Error::NotImplemented`] if this backend does not support resumable uploads. The + /// default implementation returns this, which is unreachable through the API because a + /// backend that declines in [`Self::create_upload_session`] never hands out a session. + /// - [`Error::UploadOffsetMismatch`] if `offset` is not the offset the backend holds. + /// - [`Error::UploadSessionGone`] if the session expired or was terminated. + /// - [`Error::InvalidUploadRequest`] if the session is unusable, or the chunk would + /// exceed the length declared at creation. + async fn put_chunk( + &self, + id: &ObjectId, + session: &SessionToken, + offset: u64, + content_length: u64, + stream: ClientStream, + ) -> Result { + let _ = (id, session, offset, content_length, stream); + Err(Error::NotImplemented) + } + + /// Reports how far the session has progressed, committing the object if it is assembled. + /// + /// This is the recovery path: after any failed chunk the client calls this and continues + /// from the returned offset. It is also the only read-shaped operation that mutates + /// state. Making an object visible can outlive the request that triggered it, so a + /// session whose payload fully landed may still be uncommitted; this operation finishes + /// that work and returns [`UploadProgress::Committed`]. Callers must therefore treat it + /// as a write. + /// + /// A session whose object was assembled but not yet committed never reports + /// [`UploadProgress::Committed`], so a client that observes completion can always read + /// the object back. + /// + /// # Errors + /// + /// The same conditions as [`Self::put_chunk`], except for the offset mismatch. + async fn upload_offset(&self, id: &ObjectId, session: &SessionToken) -> Result { + let _ = (id, session); + Err(Error::NotImplemented) + } + + /// Terminates a session, discarding whatever was uploaded. + /// + /// Idempotent. Not required for correctness, since sessions expire on their own, but it + /// lets a caller release an abandoned upload immediately. + /// + /// # Errors + /// + /// - [`Error::NotImplemented`] if this backend does not support resumable uploads. + /// - [`Error::InvalidUploadRequest`] if the session token is unusable. + async fn terminate_upload( + &self, + id: &ObjectId, + session: &SessionToken, + ) -> Result { + let _ = (id, session); + Err(Error::NotImplemented) + } } /// Trait for backends that support our S3-style multipart upload protocol. diff --git a/objectstore-service/src/backend/counting.rs b/objectstore-service/src/backend/counting.rs index 4e597e0f..7f8b25f3 100644 --- a/objectstore-service/src/backend/counting.rs +++ b/objectstore-service/src/backend/counting.rs @@ -28,6 +28,9 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::{ + CreateSessionResponse, SessionToken, TerminateUploadResponse, UploadProgress, +}; use crate::stream::ClientStream; /// Increments `cogs.usage` by one operation for the given `usecase`. @@ -46,6 +49,12 @@ fn count(usecase: &str) { /// `Arc`s that point to the inner backend: /// - `inner: Arc` /// - `inner_multipart: Option>` if `inner` supports it +/// +/// Resumable uploads avoid this problem: their operations live on [`Backend`] itself and express +/// support by declining in +/// [`create_upload_session`](Backend::create_upload_session), so this decorator forwards them like +/// any other method. Forwarding is mandatory — without it the decorator's declining default would +/// shadow an inner backend that does support resumable uploads. #[derive(Debug)] pub struct CountingBackend { inner: Arc, @@ -99,6 +108,46 @@ impl Backend for CountingBackend { self.inner.as_multipart_upload_backend()?; Ok(self) } + + async fn create_upload_session( + &self, + id: &ObjectId, + metadata: &Metadata, + total_length: u64, + ) -> Result { + count(&id.context.usecase); + self.inner + .create_upload_session(id, metadata, total_length) + .await + } + + async fn put_chunk( + &self, + id: &ObjectId, + session: &SessionToken, + offset: u64, + content_length: u64, + stream: ClientStream, + ) -> Result { + count(&id.context.usecase); + self.inner + .put_chunk(id, session, offset, content_length, stream) + .await + } + + async fn upload_offset(&self, id: &ObjectId, session: &SessionToken) -> Result { + count(&id.context.usecase); + self.inner.upload_offset(id, session).await + } + + async fn terminate_upload( + &self, + id: &ObjectId, + session: &SessionToken, + ) -> Result { + count(&id.context.usecase); + self.inner.terminate_upload(id, session).await + } } #[async_trait::async_trait] diff --git a/objectstore-service/src/backend/testing.rs b/objectstore-service/src/backend/testing.rs index 24034f24..71e56250 100644 --- a/objectstore-service/src/backend/testing.rs +++ b/objectstore-service/src/backend/testing.rs @@ -52,6 +52,9 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::{ + CreateSessionResponse, SessionToken, TerminateUploadResponse, UploadProgress, +}; use crate::stream::ClientStream; /// Hooks for [`TestBackend`]. @@ -237,6 +240,60 @@ pub trait Hooks: fmt::Debug + Send + Sync + 'static { ) -> Result { inner.complete_multipart(id, upload_id, parts).await } + + // --- Resumable upload methods --- + // + // `InMemoryBackend` does not implement resumable uploads, so these delegate to the + // declining `Backend` defaults. A test that exercises the resumable protocol has to + // override them. + + /// Intercepts [`Backend::create_upload_session`]. Default delegates to `inner`. + async fn create_upload_session( + &self, + inner: &InMemoryBackend, + id: &ObjectId, + metadata: &Metadata, + total_length: u64, + ) -> Result { + inner + .create_upload_session(id, metadata, total_length) + .await + } + + /// Intercepts [`Backend::put_chunk`]. Default delegates to `inner`. + async fn put_chunk( + &self, + inner: &InMemoryBackend, + id: &ObjectId, + session: &SessionToken, + offset: u64, + content_length: u64, + stream: ClientStream, + ) -> Result { + inner + .put_chunk(id, session, offset, content_length, stream) + .await + } + + /// Intercepts [`Backend::upload_offset`]. Default delegates to `inner`. + async fn upload_offset( + &self, + inner: &InMemoryBackend, + id: &ObjectId, + session: &SessionToken, + ) -> Result { + inner.upload_offset(id, session).await + } + + /// Intercepts [`Backend::terminate_upload`]. Default delegates to `inner`. + async fn terminate_upload( + &self, + inner: &InMemoryBackend, + id: &ObjectId, + session: &SessionToken, + ) -> Result { + inner.terminate_upload(id, session).await + } } /// Generic test backend that implements both [`Backend`] and [`HighVolumeBackend`]. @@ -311,6 +368,42 @@ impl Backend for TestBackend { async fn join(&self) { self.hooks.join(&self.inner).await } + + async fn create_upload_session( + &self, + id: &ObjectId, + metadata: &Metadata, + total_length: u64, + ) -> Result { + self.hooks + .create_upload_session(&self.inner, id, metadata, total_length) + .await + } + + async fn put_chunk( + &self, + id: &ObjectId, + session: &SessionToken, + offset: u64, + content_length: u64, + stream: ClientStream, + ) -> Result { + self.hooks + .put_chunk(&self.inner, id, session, offset, content_length, stream) + .await + } + + async fn upload_offset(&self, id: &ObjectId, session: &SessionToken) -> Result { + self.hooks.upload_offset(&self.inner, id, session).await + } + + async fn terminate_upload( + &self, + id: &ObjectId, + session: &SessionToken, + ) -> Result { + self.hooks.terminate_upload(&self.inner, id, session).await + } } #[async_trait::async_trait] diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index a6af1342..43c0e96d 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/objectstore-service/src/backend/tiered.rs @@ -96,6 +96,16 @@ //! already-mutated state and still returns `true` — so callers do not mistakenly //! treat a successful commit as a lost race and clean up data that was actually //! persisted. +//! +//! # Resumable Uploads +//! +//! Not implemented here yet, so [`TieredStorage`] inherits the declining defaults from +//! [`Backend`] and every session creation is denied. A resumable upload will be a regular +//! long-term write whose payload arrives across several requests, reusing the revision keys, +//! changelog phases and compare-and-write commit described above: session creation decides +//! the tier from the declared total length and declines if that tier cannot support it, +//! non-final chunks pass straight through to the upstream session, and the final chunk runs +//! the long-term write sequence. use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; diff --git a/objectstore-service/src/error.rs b/objectstore-service/src/error.rs index 9643dd43..678cf94b 100644 --- a/objectstore-service/src/error.rs +++ b/objectstore-service/src/error.rs @@ -151,6 +151,29 @@ pub enum Error { /// Invalid upload ID (e.g. path traversal attempt). #[error(transparent)] InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId), + + /// A resumable chunk was submitted at an offset the backend does not hold. + /// + /// The client resynchronizes by continuing from [`offset`](Self::UploadOffsetMismatch::offset), + /// which is authoritative and may be lower than the end of a previously acknowledged chunk. + #[error("upload offset mismatch (server holds {offset} bytes)")] + UploadOffsetMismatch { + /// The offset the backend currently holds. + offset: u64, + }, + + /// The resumable upload session expired or was terminated, retaining nothing. + /// + /// The client has to start a new session. + #[error("upload session gone")] + UploadSessionGone, + + /// A resumable upload request is unusable for the session it addresses. + /// + /// Covers an unparseable or unknown session token and a chunk that would exceed the + /// length declared when the session was created. + #[error("invalid upload request: {0}")] + InvalidUploadRequest(String), } impl Error { @@ -197,6 +220,9 @@ impl Error { Self::Client(_) => Level::DEBUG, Self::Metadata(_) => Level::DEBUG, Self::RangeNotSatisfiable { .. } => Level::DEBUG, + Self::UploadOffsetMismatch { .. } => Level::DEBUG, + Self::UploadSessionGone => Level::DEBUG, + Self::InvalidUploadRequest(_) => Level::DEBUG, // Like rate limits, we treat capacity errors as warnings Self::AtCapacity => Level::WARN, // All other errors are service or backend failures diff --git a/objectstore-service/src/lib.rs b/objectstore-service/src/lib.rs index e33a2f7c..d1a2d364 100644 --- a/objectstore-service/src/lib.rs +++ b/objectstore-service/src/lib.rs @@ -8,6 +8,7 @@ pub mod error; mod gcp_auth; pub mod id; pub mod multipart; +pub mod resumable; pub mod service; pub mod stream; pub mod streaming; diff --git a/objectstore-service/src/resumable.rs b/objectstore-service/src/resumable.rs new file mode 100644 index 00000000..3039e2b9 --- /dev/null +++ b/objectstore-service/src/resumable.rs @@ -0,0 +1,46 @@ +//! Shared types for Objectstore's resumable upload protocol. +//! +//! A resumable upload is a regular write whose payload arrives across several requests. +//! A session declares the object's total size and metadata upfront; chunks then arrive at +//! increasing byte offsets, and the backend commits the object itself once the last byte +//! lands. See [`objectstore_types::resumable`] for the wire-level types. +//! +//! Not every backend can support this. Session creation therefore asks the backend that +//! would store the object to open one, and a backend that cannot declines by returning +//! `None` from +//! [`Backend::create_upload_session`](crate::backend::common::Backend::create_upload_session). +//! Declining is a routine outcome rather than an error: the server denies the session and +//! the client falls back to a regular upload. + +pub use objectstore_types::resumable::{InvalidSessionToken, SessionToken, UploadOffset}; + +/// How far a resumable upload has progressed. +/// +/// Returned by both +/// [`Backend::put_chunk`](crate::backend::common::Backend::put_chunk) and +/// [`Backend::upload_offset`](crate::backend::common::Backend::upload_offset), because an +/// offset query commits an object that was assembled but not yet committed and therefore +/// has the same two outcomes as a chunk write. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum UploadProgress { + /// More bytes are expected. The client continues from `offset`. + /// + /// This offset is authoritative and may be lower than the end of the chunk that was + /// just written: backends persist only aligned prefixes and discard the remainder. + Incomplete { + /// The offset the backend has persisted. + offset: u64, + }, + /// The last byte arrived and the object is committed and readable. + Committed, +} + +/// Response for +/// [`Backend::create_upload_session`](crate::backend::common::Backend::create_upload_session). +/// +/// `None` means the backend declines resumable uploads for this object. +pub type CreateSessionResponse = Option; + +/// Response for +/// [`Backend::terminate_upload`](crate::backend::common::Backend::terminate_upload). +pub type TerminateUploadResponse = (); diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 0953fcd8..ca8dc68c 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -20,6 +20,9 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::{ + CreateSessionResponse, SessionToken, TerminateUploadResponse, UploadProgress, +}; use crate::stream::{ClientStream, PayloadStream}; use crate::streaming::StreamExecutor; @@ -359,6 +362,82 @@ impl StorageService { }) .await } + + // --- Resumable upload operations --- + + /// Opens a resumable upload session for an object of `total_length` bytes. + /// + /// Returns `Ok(None)` when the backend declines resumable uploads for this object, in + /// which case the caller should fall back to [`Self::insert_object`]. Unlike the + /// multipart operations there is no eager capability probe: support is the return value. + pub async fn create_upload_session( + &self, + id: ObjectId, + metadata: Metadata, + total_length: u64, + ) -> Result { + metadata.validate()?; + let inner = Arc::clone(&self.inner); + self.spawn("create_upload_session", async move { + inner + .create_upload_session(&id, &metadata, total_length) + .await + }) + .await + } + + /// Writes a chunk of `content_length` bytes at `offset` into an open session. + /// + /// Commits the object once the chunk carrying the last byte is persisted. + /// + /// # Run-to-completion + /// + /// Once called, the operation runs to completion even if the returned future is dropped. + /// This matters most for the final chunk, which commits the object. + pub async fn put_chunk( + &self, + id: ObjectId, + session: SessionToken, + offset: u64, + content_length: u64, + body: ClientStream, + ) -> Result { + let inner = Arc::clone(&self.inner); + self.spawn("put_chunk", async move { + inner + .put_chunk(&id, &session, offset, content_length, body) + .await + }) + .await + } + + /// Reports how far a session has progressed, committing the object if it is assembled. + /// + /// This mutates state and therefore requires write permission at the API layer. + pub async fn upload_offset( + &self, + id: ObjectId, + session: SessionToken, + ) -> Result { + let inner = Arc::clone(&self.inner); + self.spawn("upload_offset", async move { + inner.upload_offset(&id, &session).await + }) + .await + } + + /// Terminates a session, discarding whatever was uploaded. + pub async fn terminate_upload( + &self, + id: ObjectId, + session: SessionToken, + ) -> Result { + let inner = Arc::clone(&self.inner); + self.spawn("terminate_upload", async move { + inner.terminate_upload(&id, &session).await + }) + .await + } } #[cfg(test)] @@ -368,7 +447,7 @@ mod tests { use bytes::BytesMut; use futures_util::TryStreamExt; - use objectstore_types::metadata::Metadata; + use objectstore_types::metadata::{ExpirationPolicy, Metadata}; use objectstore_types::range::ByteRange; use objectstore_types::scope::{Scope, Scopes}; @@ -781,4 +860,147 @@ mod tests { "permit was not released after panic" ); } + + // --- Resumable uploads --- + + #[tokio::test] + async fn resumable_declines_by_default() { + let service = make_service(); + let id = ObjectId::new(make_context(), "resumable".into()); + let session = SessionToken::new("session".into()).unwrap(); + + let denied = service + .create_upload_session(id.clone(), Metadata::default(), 1024) + .await + .unwrap(); + assert!(denied.is_none(), "expected the backend to decline"); + + // Without a session no other operation is reachable through the API, but the + // declining defaults must still be wired up rather than panicking. + let chunk = service + .put_chunk(id.clone(), session.clone(), 0, 4, stream::single("data")) + .await; + assert!(matches!(chunk, Err(Error::NotImplemented))); + let offset = service.upload_offset(id.clone(), session.clone()).await; + assert!(matches!(offset, Err(Error::NotImplemented))); + let terminated = service.terminate_upload(id, session).await; + assert!(matches!(terminated, Err(Error::NotImplemented))); + } + + #[tokio::test] + async fn resumable_create_validates_metadata() { + let service = make_service(); + let id = ObjectId::new(make_context(), "resumable".into()); + + // A timeout policy with no resolved `time_expires` is rejected before the backend + // is consulted, exactly as it is for a regular insert. + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(60)), + ..Default::default() + }; + + let result = service.create_upload_session(id, metadata, 1024).await; + assert!(matches!(result, Err(Error::Metadata(_))), "{result:?}"); + } + + /// Backend that accepts resumable uploads and reports a fixed progression. + /// + /// Records nothing: it exists to prove that [`StorageService`] forwards arguments and + /// returns backend outcomes untouched. + #[derive(Clone, Debug)] + struct AcceptResumable { + progress: UploadProgress, + } + + #[async_trait::async_trait] + impl Hooks for AcceptResumable { + async fn create_upload_session( + &self, + _inner: &InMemoryBackend, + _id: &ObjectId, + _metadata: &Metadata, + total_length: u64, + ) -> Result { + Ok(Some( + SessionToken::new(format!("session-{total_length}")).unwrap(), + )) + } + + async fn put_chunk( + &self, + _inner: &InMemoryBackend, + _id: &ObjectId, + _session: &SessionToken, + _offset: u64, + _content_length: u64, + _stream: ClientStream, + ) -> Result { + Ok(self.progress) + } + + async fn upload_offset( + &self, + _inner: &InMemoryBackend, + _id: &ObjectId, + _session: &SessionToken, + ) -> Result { + Ok(self.progress) + } + + async fn terminate_upload( + &self, + _inner: &InMemoryBackend, + _id: &ObjectId, + _session: &SessionToken, + ) -> Result { + Ok(()) + } + } + + fn resumable_service(progress: UploadProgress) -> StorageService { + StorageService::new(Box::new(TestBackend::new(AcceptResumable { progress }))) + } + + #[tokio::test] + async fn resumable_reports_incomplete_progress() { + let service = resumable_service(UploadProgress::Incomplete { offset: 262_144 }); + let id = ObjectId::new(make_context(), "resumable".into()); + + let session = service + .create_upload_session(id.clone(), Metadata::default(), 1024) + .await + .unwrap() + .expect("session was declined"); + assert_eq!(session.as_str(), "session-1024"); + + let progress = service + .put_chunk(id.clone(), session.clone(), 0, 4, stream::single("data")) + .await + .unwrap(); + assert_eq!(progress, UploadProgress::Incomplete { offset: 262_144 }); + + let progress = service.upload_offset(id.clone(), session.clone()).await; + assert_eq!( + progress.unwrap(), + UploadProgress::Incomplete { offset: 262_144 } + ); + + service.terminate_upload(id, session).await.unwrap(); + } + + #[tokio::test] + async fn resumable_reports_commit() { + let service = resumable_service(UploadProgress::Committed); + let id = ObjectId::new(make_context(), "resumable".into()); + let session = SessionToken::new("session".into()).unwrap(); + + let progress = service + .put_chunk(id.clone(), session.clone(), 0, 4, stream::single("data")) + .await + .unwrap(); + assert_eq!(progress, UploadProgress::Committed); + + let progress = service.upload_offset(id, session).await.unwrap(); + assert_eq!(progress, UploadProgress::Committed); + } } diff --git a/objectstore-types/src/lib.rs b/objectstore-types/src/lib.rs index 910f293e..79ce8870 100644 --- a/objectstore-types/src/lib.rs +++ b/objectstore-types/src/lib.rs @@ -11,4 +11,5 @@ pub mod metadata; pub mod multipart; pub mod presign; pub mod range; +pub mod resumable; pub mod scope; diff --git a/objectstore-types/src/resumable.rs b/objectstore-types/src/resumable.rs new file mode 100644 index 00000000..a346d6e3 --- /dev/null +++ b/objectstore-types/src/resumable.rs @@ -0,0 +1,231 @@ +//! Types for the resumable upload protocol. +//! +//! A resumable upload declares the object's total size and metadata upfront, then +//! sends the payload as a sequence of chunks at increasing byte offsets. If a chunk +//! fails, the client asks the server which offset it holds and continues from there. +//! There is no finalize request: the server knows the total length from the session, +//! so it recognizes the chunk carrying the last byte and commits the object itself. +//! +//! Every request addresses the regular object endpoints with the session in the query +//! string. Header names are borrowed from [TUS] where they fit, but this is not a TUS +//! implementation: there is no version negotiation, no capability discovery, and no +//! support for uploads of unknown length. +//! +//! Key types: +//! - [`SessionToken`] — opaque identifier for an in-progress upload session. +//! - [`UploadOffset`] — the value of the [`HEADER_UPLOAD_OFFSET`] header. +//! - [`CreateSessionResponse`] — returned when a new session is created. +//! - [`CommitResponse`] — returned by the request that commits the object. +//! +//! [TUS]: https://tus.io/protocols/resumable-upload + +use std::fmt; +use std::ops::Deref; +use std::path::{Component, Path}; +use std::str::FromStr; + +use serde::{Deserialize, Deserializer, Serialize}; + +/// Request header declaring the total size of the object, in bytes. +/// +/// Required when creating a session. The server needs the total size to select a +/// backend and to recognize the final chunk. +pub const HEADER_UPLOAD_LENGTH: &str = "upload-length"; + +/// Header carrying the byte offset of a chunk, or the offset the server holds. +/// +/// On a request this is the offset of the chunk's first byte, or `*` to query the +/// server's authoritative offset. On a response it is the offset the server has +/// persisted. See [`UploadOffset`]. +pub const HEADER_UPLOAD_OFFSET: &str = "upload-offset"; + +/// The wildcard [`HEADER_UPLOAD_OFFSET`] value that queries the server's offset. +const OFFSET_WILDCARD: &str = "*"; + +/// Identifier for an in-progress resumable upload session. +/// +/// The token is opaque to the client: it is minted by the storage backend and carries +/// whatever that backend needs to continue or commit the upload without shared state. +/// It is neither signed nor encrypted, which is safe because the usecase, scopes and +/// key travel in the request path rather than in the token, so a request cannot address +/// an object other than the one it names. +/// +/// Validated on construction: non-empty and free of path-traversal components (`..`, +/// leading `/`, etc.), so a backend can safely use it as a single path segment. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize)] +#[serde(transparent)] +pub struct SessionToken(String); + +/// Error returned when a [`SessionToken`] fails validation. +#[derive(Debug, thiserror::Error)] +#[error("invalid session token: {0}")] +pub struct InvalidSessionToken(String); + +impl SessionToken { + /// Creates a new `SessionToken` after validating the input. + /// + /// # Errors + /// + /// Returns [`InvalidSessionToken`] if the string is empty or contains a component + /// that is not a plain path segment. + pub fn new(s: String) -> Result { + if s.is_empty() { + return Err(InvalidSessionToken("must not be empty".into())); + } + for component in Path::new(&s).components() { + if !matches!(component, Component::Normal(_)) { + return Err(InvalidSessionToken(s)); + } + } + Ok(Self(s)) + } + + /// Returns the session token as a string slice. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl Deref for SessionToken { + type Target = str; + + fn deref(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for SessionToken { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + +impl<'de> Deserialize<'de> for SessionToken { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + Self::new(s).map_err(serde::de::Error::custom) + } +} + +/// The value of the [`HEADER_UPLOAD_OFFSET`] request header. +/// +/// A concrete offset submits a chunk starting at that byte. The wildcard `*` submits +/// no payload and instead asks the server which offset it holds, which is also the +/// request that commits an object that was assembled but not yet committed. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum UploadOffset { + /// `Upload-Offset: ` — a chunk whose first byte sits at this offset. + At(u64), + /// `Upload-Offset: *` — a query for the server's authoritative offset. + Unknown, +} + +/// Error returned when an [`UploadOffset`] header value cannot be parsed. +#[derive(Debug, thiserror::Error)] +#[error("invalid {HEADER_UPLOAD_OFFSET} value: {0}")] +pub struct InvalidUploadOffset(String); + +impl FromStr for UploadOffset { + type Err = InvalidUploadOffset; + + fn from_str(s: &str) -> Result { + if s == OFFSET_WILDCARD { + return Ok(Self::Unknown); + } + + // Rejects the `+` sign and leading whitespace that `u64::from_str` would + // otherwise be lenient about, keeping the header canonical. + if !s.bytes().all(|b| b.is_ascii_digit()) { + return Err(InvalidUploadOffset(s.to_owned())); + } + + let offset = s.parse().map_err(|_| InvalidUploadOffset(s.to_owned()))?; + Ok(Self::At(offset)) + } +} + +impl fmt::Display for UploadOffset { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::At(offset) => offset.fmt(f), + Self::Unknown => f.write_str(OFFSET_WILDCARD), + } + } +} + +/// Response from creating a resumable upload session. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CreateSessionResponse { + /// The object key (server-generated or client-provided). + pub key: String, + /// The session token for subsequent requests. + pub session: SessionToken, +} + +/// Response from the request that commits the object. +/// +/// This is either the chunk carrying the last byte, or an offset query against a +/// session whose object was already assembled. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CommitResponse { + /// The object key. + pub key: String, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn session_token_accepts_opaque_values() -> Result<(), InvalidSessionToken> { + assert_eq!(SessionToken::new("abc123".into())?.as_str(), "abc123"); + assert_eq!( + SessionToken::new("eyJyZXZpc2lvbiI6ImEifQ".into())?.as_str(), + "eyJyZXZpc2lvbiI6ImEifQ" + ); + Ok(()) + } + + #[test] + fn session_token_rejects_empty_and_traversal() { + for invalid in ["", "..", "/abs", "a/../b", "./a"] { + assert!( + SessionToken::new(invalid.into()).is_err(), + "expected {invalid:?} to be rejected" + ); + } + } + + #[test] + fn upload_offset_parses_wildcard_and_offsets() -> Result<(), InvalidUploadOffset> { + assert_eq!("*".parse::()?, UploadOffset::Unknown); + assert_eq!("0".parse::()?, UploadOffset::At(0)); + assert_eq!("262144".parse::()?, UploadOffset::At(262144)); + Ok(()) + } + + #[test] + fn upload_offset_rejects_malformed_values() { + for invalid in ["", "-1", "+1", " 1", "1 ", "1.5", "0x10", "**", "abc"] { + assert!( + invalid.parse::().is_err(), + "expected {invalid:?} to be rejected" + ); + } + } + + #[test] + fn upload_offset_round_trips_through_display() -> Result<(), InvalidUploadOffset> { + for offset in [ + UploadOffset::Unknown, + UploadOffset::At(0), + UploadOffset::At(7), + ] { + assert_eq!(offset.to_string().parse::()?, offset); + } + Ok(()) + } +}