Encode backend into a new ObjectKey (#63)

The new `ObjectKey` encodes the used backend along with a (v7) uuid.

This can be used along with the `ScopedKey`, which also carries the
usecase and scope, to be encoded as a path.

Author: Swatinem

Cargo.lock

@@ -1,12 +1,13 @@
 use std::collections::BTreeSet;
+use std::str::FromStr;
 
 use axum::extract::FromRequestParts;
 use axum::http::request::Parts;
 use axum::http::{StatusCode, header};
 use axum::response::{IntoResponse, Response};
 use jsonwebtoken::errors::Result as JwtResult;
 use jsonwebtoken::{DecodingKey, Validation, decode};
-use objectstore_service::ObjectKey;
+use objectstore_service::{ObjectKey, ScopedKey};
 use objectstore_types::Scope;
 use serde::Deserialize;
 
@@ -28,12 +29,13 @@ pub struct Claim {
 }
 
 impl Claim {
-    pub fn into_key(self, key: String) -> ObjectKey {
-        ObjectKey {
+    pub fn into_key(self, key: String) -> anyhow::Result<ScopedKey> {
+        let key = ObjectKey::from_str(&key)?;
+        Ok(ScopedKey {
             usecase: self.usecase,
             scope: self.scope,
             key,
-        }
+        })
     }
 
     #[allow(clippy::result_large_err)]

objectstore-server/src/authentication.rs

@@ -14,7 +14,7 @@ use axum::routing::{get, put};
 use axum::{Json, Router};
 use axum_extra::middleware::option_layer;
 use futures_util::{StreamExt, TryStreamExt};
-use objectstore_service::{ObjectKey, StorageService};
+use objectstore_service::{ScopedKey, StorageService};
 use objectstore_types::Metadata;
 use sentry::integrations::tower as sentry_tower;
 use serde::Serialize;
@@ -39,8 +39,8 @@ fn make_app(state: ServiceState) -> axum::Router {
         .layer(TraceLayer::new_for_http().on_failure(DefaultOnFailure::new().level(Level::DEBUG)));
 
     let routes = Router::new()
-        .route("/", put(put_blob))
-        .route("/{*key}", get(get_blob).delete(delete_blob));
+        .route("/", put(put_object))
+        .route("/{*key}", get(get_object).delete(delete_object));
 
     routes.layer(middleware).with_state(state)
 }
@@ -101,30 +101,34 @@ struct PutBlobResponse {
 }
 
 #[tracing::instrument(level = "trace", skip(state, body))]
-async fn put_blob(
+async fn put_object(
     State(state): State<ServiceState>,
     ExtractScope(claim): ExtractScope,
     headers: HeaderMap,
     body: Body,
 ) -> error::Result<impl IntoResponse> {
     claim.ensure_permission(Permission::Write)?;
-    let key = claim.into_key(Uuid::new_v4().to_string());
     let metadata = Metadata::from_headers(&headers, "")?;
 
     let stream = body.into_data_stream().map_err(io::Error::other).boxed();
-    state.service.put_object(&key, &metadata, stream).await?;
+    let key = state
+        .service
+        .put_object(claim.usecase, claim.scope, &metadata, stream)
+        .await?;
 
-    Ok(Json(PutBlobResponse { key: key.key }))
+    Ok(Json(PutBlobResponse {
+        key: key.key.to_string(),
+    }))
 }
 
 #[tracing::instrument(level = "trace", skip(state))]
-async fn get_blob(
+async fn get_object(
     State(state): State<ServiceState>,
     ExtractScope(claim): ExtractScope,
     Path(key): Path<String>,
 ) -> error::Result<Response> {
     claim.ensure_permission(Permission::Read)?;
-    let key = claim.into_key(key);
+    let key = claim.into_key(key)?;
 
     let Some((metadata, stream)) = state.service.get_object(&key).await? else {
         return Ok(StatusCode::NOT_FOUND.into_response());
@@ -134,14 +138,14 @@ async fn get_blob(
     Ok((headers, Body::from_stream(stream)).into_response())
 }
 
-#[tracing::instrument(level = "trace", skip_all, fields(usecase, scope, key))]
-async fn delete_blob(
+#[tracing::instrument(level = "trace", skip(state))]
+async fn delete_object(
     State(state): State<ServiceState>,
     ExtractScope(claim): ExtractScope,
     Path(key): Path<String>,
 ) -> error::Result<impl IntoResponse> {
     claim.ensure_permission(Permission::Write)?;
-    let key = claim.into_key(key);
+    let key = claim.into_key(key)?;
 
     state.service.delete_object(&key).await?;
 

objectstore-server/src/http.rs

@@ -10,6 +10,7 @@ async-trait = "0.1.88"
 bigtable_rs = { git = "https://github.com/getsentry/bigtable_rs.git", branch = "admin" }
 bincode = { version = "2.0.1", features = ["serde"] }
 bytes = "1.10.1"
+data-encoding = "2.9.0"
 futures-util = "0.3.31"
 gcp_auth = "0.12.3"
 objectstore-types = { path = "../objectstore-types" }
@@ -18,6 +19,8 @@ serde = { version = "1.0.219", features = ["derive"] }
 tokio = { version = "1.46.0", features = ["full"] }
 tokio-util = "0.7.15"
 tonic = "0.13.1"
+uuid = { version = "1.18.0", features = ["v7"] }
+watto = "0.2.0"
 zstd = "0.13.3"
 
 [dev-dependencies]

objectstore-service/Cargo.toml

@@ -18,6 +18,7 @@ use objectstore_types::{ExpirationPolicy, Metadata};
 use tokio::runtime::Handle;
 use tonic::Code;
 
+use crate::ScopedKey;
 use crate::backend::{Backend, BackendStream};
 
 /// Connection timeout used for the initial connection to BigQuery.
@@ -168,13 +169,13 @@ impl BigTableBackend {
         }
     }
 
-    async fn mutate<I>(&self, path: &str, mutations: I) -> Result<MutateRowResponse>
+    async fn mutate<I>(&self, path: Vec<u8>, mutations: I) -> Result<MutateRowResponse>
     where
         I: IntoIterator<Item = Mutation>,
     {
         let request = v2::MutateRowRequest {
             table_name: self.table_path.clone(),
-            row_key: path.as_bytes().to_vec(),
+            row_key: path,
             mutations: mutations
                 .into_iter()
                 .map(|m| v2::Mutation { mutation: Some(m) })
@@ -206,10 +207,11 @@ impl BigTableBackend {}
 impl Backend for BigTableBackend {
     async fn put_object(
         &self,
-        path: &str,
+        key: &ScopedKey,
         metadata: &Metadata,
         mut stream: BackendStream,
     ) -> Result<()> {
+        let path = key.as_path().to_string().into_bytes();
         // TODO: Inject the access time from the request.
         let access_time = SystemTime::now();
 
@@ -253,9 +255,10 @@ impl Backend for BigTableBackend {
         Ok(())
     }
 
-    async fn get_object(&self, path: &str) -> Result<Option<(Metadata, BackendStream)>> {
+    async fn get_object(&self, key: &ScopedKey) -> Result<Option<(Metadata, BackendStream)>> {
+        let path = key.as_path().to_string().into_bytes();
         let rows = v2::RowSet {
-            row_keys: vec![path.as_bytes().to_vec()],
+            row_keys: vec![path.clone()],
             row_ranges: vec![],
         };
 
@@ -270,11 +273,11 @@ impl Backend for BigTableBackend {
         let response = client.read_rows(request).await?;
         debug_assert!(response.len() <= 1, "Expected at most one row");
 
-        let Some((key, cells)) = response.into_iter().next() else {
+        let Some((read_path, cells)) = response.into_iter().next() else {
             return Ok(None);
         };
 
-        debug_assert!(key == path.as_bytes(), "Row key mismatch");
+        debug_assert!(read_path == path, "Row key mismatch");
         let mut value = Bytes::new();
         let mut metadata = Metadata::default();
         let mut expire_at = None;
@@ -310,15 +313,16 @@ impl Backend for BigTableBackend {
                 let value = Bytes::clone(&value);
                 let stream = stream::once(async { Ok(value) }).boxed();
                 // TODO: Avoid the serialize roundtrip for metadata
-                self.put_object(path, &metadata, stream).await?;
+                self.put_object(key, &metadata, stream).await?;
             }
         }
 
         let stream = stream::once(async { Ok(value) }).boxed();
         Ok(Some((metadata, stream)))
     }
 
-    async fn delete_object(&self, path: &str) -> Result<()> {
+    async fn delete_object(&self, key: &ScopedKey) -> Result<()> {
+        let path = key.as_path().to_string().into_bytes();
         self.mutate(path, [Mutation::DeleteFromRow(DeleteFromRow {})])
             .await?;
         Ok(())
@@ -350,6 +354,10 @@ fn micros_to_time(micros: i64) -> Option<SystemTime> {
 mod tests {
     use std::collections::BTreeMap;
 
+    use objectstore_types::Scope;
+
+    use crate::ObjectKey;
+
     use super::*;
 
     // NB: Not run any of these tests, you need to have a BigTable emulator running and
@@ -380,15 +388,22 @@ mod tests {
         Ok(payload)
     }
 
-    fn make_path() -> String {
-        format!("usecase1/4711/{}", uuid::Uuid::new_v4())
+    fn make_key() -> ScopedKey {
+        ScopedKey {
+            usecase: "testing".into(),
+            scope: Scope {
+                organization: 1234,
+                project: Some(1234),
+            },
+            key: ObjectKey::for_backend(0),
+        }
     }
 
     #[tokio::test]
     async fn test_roundtrip() -> Result<()> {
         let backend = create_test_backend().await?;
 
-        let path = make_path();
+        let path = make_key();
         let metadata = Metadata {
             expiration_policy: ExpirationPolicy::Manual,
             compression: None,
@@ -413,7 +428,7 @@ mod tests {
     async fn test_get_nonexistent() -> Result<()> {
         let backend = create_test_backend().await?;
 
-        let path = make_path();
+        let path = make_key();
         let result = backend.get_object(&path).await?;
         assert!(result.is_none());
 
@@ -424,7 +439,7 @@ mod tests {
     async fn test_delete_nonexistent() -> Result<()> {
         let backend = create_test_backend().await?;
 
-        let path = make_path();
+        let path = make_key();
         backend.delete_object(&path).await?;
 
         Ok(())
@@ -434,7 +449,7 @@ mod tests {
     async fn test_overwrite() -> Result<()> {
         let backend = create_test_backend().await?;
 
-        let path = make_path();
+        let path = make_key();
         let metadata = Metadata {
             expiration_policy: ExpirationPolicy::Manual,
             compression: None,
@@ -469,7 +484,7 @@ mod tests {
     async fn test_read_after_delete() -> Result<()> {
         let backend = create_test_backend().await?;
 
-        let path = make_path();
+        let path = make_key();
         let metadata = Metadata {
             expiration_policy: ExpirationPolicy::Manual,
             compression: None,
@@ -495,7 +510,7 @@ mod tests {
 
         let backend = create_test_backend().await?;
 
-        let path = make_path();
+        let path = make_key();
         let metadata = Metadata {
             expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(0)),
             compression: None,
@@ -519,7 +534,7 @@ mod tests {
 
         let backend = create_test_backend().await?;
 
-        let path = make_path();
+        let path = make_key();
         let metadata = Metadata {
             expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_secs(0)),
             compression: None,

objectstore-service/src/backend/bigtable.rs

@@ -9,6 +9,8 @@ use tokio::fs::OpenOptions;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
 use tokio_util::io::{ReaderStream, StreamReader};
 
+use crate::ScopedKey;
+
 use super::{Backend, BackendStream};
 
 const BC_CONFIG: bincode::config::Configuration = bincode::config::standard();
@@ -28,10 +30,11 @@ impl LocalFs {
 impl Backend for LocalFs {
     async fn put_object(
         &self,
-        path: &str,
+        key: &ScopedKey,
         metadata: &Metadata,
         stream: BackendStream,
     ) -> anyhow::Result<()> {
+        let path = key.as_path().to_string();
         let path = self.path.join(path);
         tokio::fs::create_dir_all(path.parent().unwrap()).await?;
         let file = OpenOptions::new()
@@ -56,7 +59,11 @@ impl Backend for LocalFs {
         Ok(())
     }
 
-    async fn get_object(&self, path: &str) -> anyhow::Result<Option<(Metadata, BackendStream)>> {
+    async fn get_object(
+        &self,
+        key: &ScopedKey,
+    ) -> anyhow::Result<Option<(Metadata, BackendStream)>> {
+        let path = key.as_path().to_string();
         let path = self.path.join(path);
         let file = match OpenOptions::new().read(true).open(path).await {
             Ok(file) => file,
@@ -101,7 +108,8 @@ impl Backend for LocalFs {
         Ok(Some((metadata, stream.boxed())))
     }
 
-    async fn delete_object(&self, path: &str) -> anyhow::Result<()> {
+    async fn delete_object(&self, key: &ScopedKey) -> anyhow::Result<()> {
+        let path = key.as_path().to_string();
         let path = self.path.join(path);
         Ok(tokio::fs::remove_file(path).await?)
     }
@@ -113,7 +121,9 @@ mod tests {
 
     use bytes::BytesMut;
     use futures_util::TryStreamExt;
-    use objectstore_types::{Compression, ExpirationPolicy};
+    use objectstore_types::{Compression, ExpirationPolicy, Scope};
+
+    use crate::ObjectKey;
 
     use super::*;
 
@@ -126,17 +136,25 @@ mod tests {
         let tempdir = tempfile::tempdir().unwrap();
         let backend = LocalFs::new(tempdir.path());
 
+        let key = ScopedKey {
+            usecase: "testing".into(),
+            scope: Scope {
+                organization: 1234,
+                project: Some(1234),
+            },
+            key: ObjectKey::for_backend(0),
+        };
         let metadata = Metadata {
             expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_secs(3600)),
             compression: Some(Compression::Zstd),
             custom: [("foo".into(), "bar".into())].into(),
         };
         backend
-            .put_object("foo/bar", &metadata, make_stream(b"oh hai!"))
+            .put_object(&key, &metadata, make_stream(b"oh hai!"))
             .await
             .unwrap();
 
-        let (read_metadata, stream) = backend.get_object("foo/bar").await.unwrap().unwrap();
+        let (read_metadata, stream) = backend.get_object(&key).await.unwrap().unwrap();
         let file_contents: BytesMut = stream.try_collect().await.unwrap();
 
         assert_eq!(read_metadata, metadata);