Skip to content

feat(server): Add resumable upload API #593

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jan-auer wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat(server): Add resumable upload API #593

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 10 additions & 43 deletions objectstore-server/docs/architecture.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
60 changes: 60 additions & 0 deletions objectstore-server/src/auth/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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<CreateSessionResponse> {
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<UploadProgress> {
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<UploadProgress> {
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<TerminateUploadResponse> {
self.check_permission(Permission::ObjectWrite, id.context())?;
Ok(self.service.terminate_upload(id, session).await?)
}
}
14 changes: 14 additions & 0 deletions objectstore-server/src/endpoints/common.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>) -> 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<E: Error + ?Sized>(error: &E) -> Self {
let detail = Some(error.to_string());
Expand Down Expand Up @@ -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,
Expand Down
109 changes: 109 additions & 0 deletions objectstore-server/src/endpoints/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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=<token>` 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=<token>` | Upload a chunk, or query the offset |
//! | `DELETE` | `/v1/objects/{usecase}/{scopes}/{*key}?session=<token>` | 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;
Expand All @@ -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 {
Expand Down
51 changes: 47 additions & 4 deletions objectstore-server/src/endpoints/objects.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -43,9 +44,27 @@ async fn objects_post(
service: AuthAwareService,
State(state): State<ServiceState>,
Xt(context): Xt<ObjectContext>,
OriginalUri(uri): OriginalUri,
Query(query): Query<ResumableQuery>,
headers: HeaderMap,
MeteredBody(body): MeteredBody,
) -> ApiResult<Response> {
// 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
Expand Down Expand Up @@ -199,9 +218,26 @@ async fn object_put(
service: AuthAwareService,
State(state): State<ServiceState>,
Xt(id): Xt<ObjectId>,
OriginalUri(uri): OriginalUri,
Query(query): Query<ResumableQuery>,
headers: HeaderMap,
MeteredBody(body): MeteredBody,
body: MeteredBody,
) -> ApiResult<Response> {
// `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;
Expand All @@ -226,7 +262,14 @@ async fn object_put(
async fn object_delete(
service: AuthAwareService,
Xt(id): Xt<ObjectId>,
) -> ApiResult<impl IntoResponse> {
Query(query): Query<ResumableQuery>,
) -> ApiResult<Response> {
// 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())
}
Loading
Loading