feat(server): Add Resumable Upload API

objectstore-server/Cargo.toml

@@ -15,6 +15,7 @@ anyhow = { workspace = true }
 argh = { workspace = true }
 async-stream = { workspace = true }
 axum = { workspace = true, features = ["multipart"] }
+base64 = { workspace = true }
 bytes = { workspace = true }
 ed25519-dalek = { workspace = true, features = ["pem"] }
 elegant-departure = { workspace = true, features = ["tokio"] }

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
 

objectstore-server/src/auth/service.rs

@@ -3,6 +3,7 @@ use objectstore_service::multipart::{
     AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
     ListPartsResponse, PartNumber, UploadId, UploadPartResponse,
 };
+use objectstore_service::resumable::{SessionToken, UploadProgress};
 use objectstore_service::service::{DeleteResponse, GetResponse, InsertResponse, MetadataResponse};
 
 use objectstore_service::{ClientStream, StorageService};
@@ -186,4 +187,57 @@ 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 canceling 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<SessionToken> {
+        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::cancel_upload`].
+    pub async fn cancel_upload(&self, id: ObjectId, session: SessionToken) -> ApiResult<()> {
+        self.check_permission(Permission::ObjectWrite, id.context())?;
+        Ok(self.service.cancel_upload(id, session).await?)
+    }
 }

objectstore-server/src/endpoints/common.rs

@@ -96,6 +96,12 @@ impl ApiError {
                 StatusCode::RANGE_NOT_SATISFIABLE
             }
             ApiError::Service(ServiceError::InvalidUploadId(_)) => StatusCode::BAD_REQUEST,
+            ApiError::Service(ServiceError::UnknownUploadSession) => StatusCode::BAD_REQUEST,
+            ApiError::Service(ServiceError::ChunkExceedsUploadLength { .. }) => {
+                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,

objectstore-server/src/endpoints/mod.rs

@@ -1,5 +1,116 @@
 //! 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 can ask the server which offset it holds and continue from there,
+//! so an interrupted transfer resumes where it stopped instead of starting over.
+//! Clients are still encouraged to send the whole payload in a single request, as that's the most
+//! efficient and reliable approach.
+//! 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.
+//! Session creation returns the opaque backend token unchanged; subsequent requests encode it as
+//! unpadded base64url in the `session` query parameter.
+//! 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>`    | Cancel upload, discarding what was sent      |
+//!
+//! Session creation requires an `Upload-Length` header carrying the total size of the object
+//! in bytes, takes the same metadata headers as a regular upload, and requires an empty body.
+//! It answers `200 OK` with `{"key", "session"}`; the session field is the token to use in
+//! subsequent query parameters. 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 last chunk that was sent.
+//! Backends can e.g. persist only aligned prefixes and discard the remainder, so clients must
+//! always continue from the returned offset.
+//! Every chunk requires `Content-Length`, even over HTTP/2, while creation and offset queries
+//! must not carry a request body.
+//!
+//! An offset query can commit an object, 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: unknown upload session, missing `Upload-Length`, nonempty offset query, or a chunk exceeding the declared length | Terminal |
+//! | `409`  | A chunk's offset does not match, with the authoritative offset in `Upload-Offset` | Resynchronize |
+//! | `410`  | The session expired or was canceled; nothing was retained | Start a new session |
+//! | `501`  | The configured backend does not implement resumable uploads | Fall back to a regular upload |
+//!
+//! # 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 +125,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 {

objectstore-server/src/endpoints/objects.rs

@@ -1,7 +1,8 @@
 use std::fmt::Write as _;
 
 use axum::body::Body;
-use axum::extract::State;
+use axum::extract::{Request, State};
+use axum::handler::Handler;
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use axum::routing;
@@ -15,31 +16,79 @@ use serde::Serialize;
 
 use crate::auth::AuthAwareService;
 use crate::endpoints::common::{ApiError, ApiResult, insert_accept_ranges};
+use crate::endpoints::resumable::{self, ResumableTarget};
 use crate::extractors::byte_range::OptionalByteRange;
 use crate::extractors::{Xt, body::MeteredBody};
 use crate::state::ServiceState;
 
 pub fn router() -> Router<ServiceState> {
-    let collection_routes = routing::post(objects_post);
+    let collection_routes = routing::post(dispatch_objects_post);
     let object_routes = routing::get(object_get)
         .head(object_head)
-        .put(object_put)
+        .put(dispatch_object_put)
         // TODO(ja): Implement PATCH (metadata update w/o body)
-        .delete(object_delete);
+        .delete(dispatch_object_delete);
 
     Router::new()
         .route("/objects/{usecase}/{scopes}", collection_routes.clone())
         .route("/objects/{usecase}/{scopes}/", collection_routes)
         .route("/objects/{usecase}/{scopes}/{*key}", object_routes)
 }
 
+async fn dispatch_objects_post(
+    State(state): State<ServiceState>,
+    target: Option<ResumableTarget>,
+    request: Request,
+) -> Response {
+    match target {
+        Some(ResumableTarget::NewSession) => resumable::create_session.call(request, state).await,
+        Some(ResumableTarget::ExistingSession) => {
+            ApiError::Client("`session` requires an object key; use PUT on the object path".into())
+                .into_response()
+        }
+        None => create_object.call(request, state).await,
+    }
+}
+
+async fn dispatch_object_put(
+    State(state): State<ServiceState>,
+    target: Option<ResumableTarget>,
+    request: Request,
+) -> Response {
+    match target {
+        Some(ResumableTarget::NewSession) => {
+            resumable::create_session_for_key.call(request, state).await
+        }
+        Some(ResumableTarget::ExistingSession) => {
+            resumable::continue_session.call(request, state).await
+        }
+        None => insert_object.call(request, state).await,
+    }
+}
+
+async fn dispatch_object_delete(
+    State(state): State<ServiceState>,
+    target: Option<ResumableTarget>,
+    request: Request,
+) -> Response {
+    match target {
+        Some(ResumableTarget::ExistingSession) => {
+            resumable::cancel_session.call(request, state).await
+        }
+        Some(ResumableTarget::NewSession) => {
+            ApiError::Client("`upload_type` is not supported on DELETE".into()).into_response()
+        }
+        None => delete_object.call(request, state).await,
+    }
+}
+
 /// Response returned when inserting an object.
 #[derive(Debug, Serialize)]
 pub struct InsertObjectResponse {
     pub key: String,
 }
 
-async fn objects_post(
+async fn create_object(
     service: AuthAwareService,
     State(state): State<ServiceState>,
     Xt(context): Xt<ObjectContext>,
@@ -195,7 +244,7 @@ fn format_content_disposition(filename: &str) -> http::HeaderValue {
     http::HeaderValue::from_str(&result).expect("content disposition is a valid header value")
 }
 
-async fn object_put(
+async fn insert_object(
     service: AuthAwareService,
     State(state): State<ServiceState>,
     Xt(id): Xt<ObjectId>,
@@ -223,7 +272,7 @@ async fn object_put(
     Ok((StatusCode::OK, response).into_response())
 }
 
-async fn object_delete(
+async fn delete_object(
     service: AuthAwareService,
     Xt(id): Xt<ObjectId>,
 ) -> ApiResult<impl IntoResponse> {