feat(gcs): Add resumable uploads

Cargo.toml

@@ -80,6 +80,7 @@ rand = "0.10.1"
 rand_distr = "0.6.0"
 regex = "1.12.4"
 reqwest = { version = "0.13.4", default-features = false }
+ring = "0.17.14"
 rustls = { version = "0.23.40", default-features = false }
 secrecy = "0.10.3"
 sentry = "0.48.3"

objectstore-server/Cargo.toml

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

objectstore-server/src/auth/service.rs

@@ -226,7 +226,7 @@ impl AuthAwareService {
         id: ObjectId,
         session: SessionToken,
     ) -> ApiResult<UploadProgress> {
-        // An offset query can commit an assembled object.
+        // A status query can detect an upload to be complete and cause logical object creation.
         self.check_permission(Permission::ObjectWrite, id.context())?;
         Ok(self.service.upload_offset(id, session).await?)
     }

objectstore-server/src/config.rs

@@ -65,6 +65,7 @@ use anyhow::Result;
 use figment::providers::{Env, Format, Serialized, Yaml};
 use objectstore_service::backend::local_fs::FileSystemConfig;
 use objectstore_service::change_stream::CostTrackerConfig;
+use objectstore_service::resumable::Encryptor;
 use objectstore_types::auth::Permission;
 use secrecy::{CloneableSecret, SecretBox, SerializableSecret, zeroize::Zeroize};
 use serde::{Deserialize, Serialize};
@@ -575,6 +576,8 @@ pub struct Config {
 /// - `OS__SERVICE__CONCURRENCY_QUEUE`
 /// - `OS__SERVICE__CONCURRENCY_TIMEOUT`
 /// - `OS__SERVICE__BULK_CONCURRENCY_PCT`
+/// - `OS__SERVICE__RESUMABLE_TOKEN_ENCRYPTION__ACTIVE_KEY_ID`
+/// - `OS__SERVICE__RESUMABLE_TOKEN_ENCRYPTION__KEY_FILES`
 #[derive(Debug, Deserialize, Serialize)]
 #[serde(default)]
 pub struct Service {
@@ -629,6 +632,61 @@ pub struct Service {
     ///
     /// `60`
     pub bulk_concurrency_pct: u32,
+
+    /// Persistent encryption keys for resumable-upload session tokens returned to clients.
+    ///
+    /// Tokens are always encrypted. When this is absent, Objectstore generates a fresh in-memory
+    /// AES-256 key at startup, so resumable sessions become invalid after a restart. Configure a
+    /// persistent keyring for sessions that must survive restarts. Keep old keys configured while
+    /// their sessions may still be active; removing a key intentionally invalidates those sessions.
+    ///
+    /// ```yaml
+    /// service:
+    ///   resumable_token_encryption:
+    ///     active_key_id: v1
+    ///     key_files:
+    ///       v1: /var/run/secrets/objectstore/resumable-upload-v1
+    /// ```
+    pub resumable_token_encryption: Option<ResumableTokenEncryptionConfig>,
+}
+
+impl Service {
+    /// Loads and validates the configured resumable token encryption keys.
+    pub(crate) fn resumable_token_encryption(&self) -> Result<Option<Encryptor>> {
+        let Some(config) = &self.resumable_token_encryption else {
+            return Ok(None);
+        };
+
+        let mut keys = BTreeMap::new();
+        for (key_id, filename) in &config.key_files {
+            let bytes = std::fs::read(filename).map_err(|error| {
+                anyhow::anyhow!("reading resumable token key {filename:?}: {error}")
+            })?;
+            keys.insert(key_id.clone(), bytes);
+        }
+
+        Encryptor::new(config.active_key_id.clone(), keys).map(Some)
+    }
+}
+
+/// AES-256-GCM keys used to protect externally visible resumable session tokens.
+#[derive(Clone, Deserialize, Serialize)]
+pub struct ResumableTokenEncryptionConfig {
+    /// Key used to encrypt newly created sessions.
+    pub active_key_id: String,
+    /// Files containing raw, exactly 32-byte AES-256 keys, indexed by rotation ID.
+    #[serde(default)]
+    pub key_files: BTreeMap<String, PathBuf>,
+}
+
+impl fmt::Debug for ResumableTokenEncryptionConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ResumableTokenEncryptionConfig")
+            .field("active_key_id", &self.active_key_id)
+            .field("key_ids", &self.key_files.keys().collect::<Vec<_>>())
+            .field("key_files", &self.key_files)
+            .finish()
+    }
 }
 
 impl Default for Service {
@@ -638,6 +696,7 @@ impl Default for Service {
             concurrency_queue: 0,
             concurrency_timeout: Duration::from_secs(1),
             bulk_concurrency_pct: 60,
+            resumable_token_encryption: None,
         }
     }
 }
@@ -866,6 +925,78 @@ mod tests {
         });
     }
 
+    #[test]
+    fn resumable_token_encryption_loads_keys_from_files() {
+        let mut key_file = tempfile::NamedTempFile::new().unwrap();
+        key_file.write_all(&[7; 32]).unwrap();
+        let mut tempfile = tempfile::NamedTempFile::new().unwrap();
+        tempfile
+            .write_all(
+                format!(
+                    "service:\n  resumable_token_encryption:\n    active_key_id: v1\n    key_files:\n      v1: \"{}\"\n",
+                    key_file.path().display(),
+                )
+                .as_bytes(),
+            )
+            .unwrap();
+
+        figment::Jail::expect_with(|_jail| {
+            let config = Config::load(Some(tempfile.path())).unwrap();
+            assert!(
+                config
+                    .service
+                    .resumable_token_encryption()
+                    .unwrap()
+                    .is_some()
+            );
+
+            let debug = format!("{:?}", config.service);
+            assert!(debug.contains("v1"));
+            assert!(debug.contains(&key_file.path().display().to_string()));
+            assert!(!debug.contains("07070707"));
+            Ok(())
+        });
+    }
+
+    #[test]
+    fn resumable_token_encryption_rejects_invalid_configuration() {
+        let mut valid = tempfile::NamedTempFile::new().unwrap();
+        valid.write_all(&[7; 32]).unwrap();
+        let mut short = tempfile::NamedTempFile::new().unwrap();
+        short.write_all(&[7; 31]).unwrap();
+        let missing = valid.path().with_extension("missing");
+        for yaml in [
+            "service:\n  resumable_token_encryption:\n    active_key_id: v1\n".to_owned(),
+            format!(
+                "service:\n  resumable_token_encryption:\n    active_key_id: missing\n    key_files:\n      v1: \"{}\"\n",
+                valid.path().display(),
+            ),
+            format!(
+                "service:\n  resumable_token_encryption:\n    active_key_id: bad_key\n    key_files:\n      'bad key': \"{}\"\n",
+                valid.path().display(),
+            ),
+            format!(
+                "service:\n  resumable_token_encryption:\n    active_key_id: v1\n    key_files:\n      v1: \"{}\"\n",
+                short.path().display(),
+            ),
+            format!(
+                "service:\n  resumable_token_encryption:\n    active_key_id: v1\n    key_files:\n      v1: \"{}\"\n",
+                missing.display(),
+            ),
+        ] {
+            let mut tempfile = tempfile::NamedTempFile::new().unwrap();
+            tempfile.write_all(yaml.as_bytes()).unwrap();
+            figment::Jail::expect_with(|_jail| {
+                let config = Config::load(Some(tempfile.path())).unwrap();
+                assert!(
+                    config.service.resumable_token_encryption().is_err(),
+                    "accepted {yaml}"
+                );
+                Ok(())
+            });
+        }
+    }
+
     #[test]
     fn configured_with_env_and_yaml() {
         let mut tempfile = tempfile::NamedTempFile::new().unwrap();

objectstore-server/src/endpoints/mod.rs

@@ -32,12 +32,12 @@
 //! 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.
+//! byte and completes the upload 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.
+//! Session creation returns an opaque token encoded once as unpadded base64url; that value can be
+//! placed directly in the `session` query parameter.
 //! The object is named by the request path as usual, and [`objectstore_types::resumable`]
 //! holds the protocol types.
 //!
@@ -57,14 +57,16 @@
 //! `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.
+//! `201 Created` with `{"key"}` once the upload is complete and the object is available through
+//! the normal object endpoints. The session is terminal at that point.
 //! 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.
+//! An offset query can finish pending backend publication work, 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.
 //!

objectstore-server/src/endpoints/resumable.rs

@@ -23,7 +23,8 @@ use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_service::stream::ClientStream;
 use objectstore_types::metadata::Metadata;
 use objectstore_types::resumable::{
-    CommitResponse, CreateSessionResponse, HEADER_UPLOAD_OFFSET, UploadOffset, UploadProgress,
+    CompleteUploadResponse, CreateSessionResponse, HEADER_UPLOAD_OFFSET, UploadOffset,
+    UploadProgress,
 };
 
 use crate::auth::AuthAwareService;
@@ -113,12 +114,17 @@ async fn create_session_for_id(
 ///
 /// [`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.
+/// server stands. A status query can observe completion when the final chunk succeeded but its
+/// response was lost. A composed backend can also finish pending idempotent publication work
+/// before reporting completion.
 /// Chunks require `Content-Length`, including over HTTP/2. Offset queries may omit it, but the
 /// body stream is checked and any bytes are rejected as a malformed request.
 ///
 /// Both answer `204 No Content` with the authoritative offset while bytes remain, and
-/// `201 Created` with the key once the object is committed.
+/// `201 Created` with the key once the upload is complete, the session is terminal, and the object
+/// is available through the normal object endpoints.
+/// The acknowledged offset may be lower than the submitted chunk's end, so clients should continue
+/// from this response (or from a later explicit offset query), never from local byte accounting alone.
 pub(super) async fn continue_session(
     service: AuthAwareService,
     Xt(id): Xt<ObjectId>,
@@ -189,8 +195,8 @@ fn progress_response(progress: ApiResult<UploadProgress>, key: String) -> ApiRes
             [(HEADER_UPLOAD_OFFSET, http::HeaderValue::from(offset))],
         )
             .into_response(),
-        UploadProgress::Committed => {
-            (StatusCode::CREATED, Json(CommitResponse { key })).into_response()
+        UploadProgress::Complete => {
+            (StatusCode::CREATED, Json(CompleteUploadResponse { key })).into_response()
         }
     };
 
@@ -229,7 +235,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_answers_created_with_the_key() {
-        let response = progress_response(Ok(UploadProgress::Committed), "my-key".into()).unwrap();
+        let response = progress_response(Ok(UploadProgress::Complete), "my-key".into()).unwrap();
 
         let (status, offset, body) = parts_of(response).await;
         assert_eq!(status, StatusCode::CREATED);

objectstore-server/src/extractors/id.rs

@@ -239,7 +239,7 @@ mod tests {
     use crate::web::RequestCounter;
 
     async fn test_state(config: Config) -> ServiceState {
-        let service = StorageService::new(Box::new(InMemoryBackend::new("in-memory")));
+        let service = StorageService::new(Box::new(InMemoryBackend::new("in-memory"))).unwrap();
         let key_directory = Arc::new(PublicKeyDirectory::from_config(&config.auth).await.unwrap());
         let rate_limiter = RateLimiter::new(config.rate_limits.clone());
 

objectstore-server/src/resumable.rs

@@ -3,8 +3,6 @@
 use axum::extract::{FromRequestParts, OptionalFromRequestParts, Query};
 use axum::http::{HeaderName, HeaderValue, request::Parts};
 use axum_extra::headers::{ContentLength, Error as HeaderError, Header};
-use base64::Engine as _;
-use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use futures_util::TryStreamExt;
 use objectstore_service::error::Error as ServiceError;
 use objectstore_service::stream::ClientStream;
@@ -177,19 +175,8 @@ where
 }
 
 fn decode_session_token(encoded: &str) -> ApiResult<SessionToken> {
-    let bytes = URL_SAFE_NO_PAD
-        .decode(encoded)
-        .map_err(|error| ApiError::map_client("invalid session token", error))?;
-
-    if URL_SAFE_NO_PAD.encode(&bytes) != encoded {
-        return Err(ApiError::client(
-            "session token must use unpadded base64 URL encoding",
-        ));
-    }
-
-    String::from_utf8(bytes)
-        .map(SessionToken::from)
-        .map_err(|error| ApiError::map_client("session token is not valid UTF-8", error))
+    SessionToken::from_base64url(encoded)
+        .map_err(|error| ApiError::map_client("invalid session token", error))
 }
 
 /// Confirms that a request neither declares nor streams a non-empty body.
@@ -255,14 +242,14 @@ mod tests {
     #[test]
     fn session_token_decodes_from_unpadded_base64url() {
         assert_eq!(
-            decode_session_token("Li4vZXNjYXBl").unwrap().as_ref(),
-            "../escape"
+            decode_session_token("Li4vZXNjYXBl").unwrap().as_bytes(),
+            b"../escape"
         );
     }
 
     #[test]
     fn session_token_rejects_invalid_query_encodings() {
-        for invalid in ["%%%", "dG9rM24=", "_w"] {
+        for invalid in ["%%%", "dG9rM24="] {
             assert!(
                 decode_session_token(invalid).is_err(),
                 "accepted {invalid:?}"

objectstore-server/src/state.rs

@@ -68,12 +68,16 @@ impl Services {
             .as_ref()
             .map(ChangeStreamFactory::new)
             .unwrap_or_default();
+        let resumable_token_encryption = config.service.resumable_token_encryption()?;
         let backend = backend::from_config(config.storage.clone(), &streams).await?;
         let concurrency = ConcurrencyLimiter::new(config.service.max_concurrency)
             .with_queue(config.service.concurrency_queue)
             .with_timeout(config.service.concurrency_timeout)
             .with_bulk(config.service.bulk_concurrency_pct);
-        let service = StorageService::new(backend).with_concurrency(concurrency);
+        let mut service = StorageService::new(backend)?.with_concurrency(concurrency);
+        if let Some(resumable_token_encryption) = resumable_token_encryption {
+            service = service.with_resumable_token_encryption(resumable_token_encryption);
+        }
         service.start();
 
         let key_directory = Arc::new(PublicKeyDirectory::from_config(&config.auth).await?);