Enhance tracing and refactor blob-related keyspaces

Cargo.toml

@@ -88,9 +88,15 @@ opendal = { version = "0.56", features = [
     "services-http",
     "services-webdav",
 ] }
-opentelemetry = "0.31.0"
-opentelemetry-otlp = { version = "0.31.1", features = ["grpc-tonic"] }
-opentelemetry_sdk = { version = "0.31.0", features = ["rt-tokio"] }
+opentelemetry = { version = "0.31.0", default-features = false, features = ["trace"] }
+opentelemetry-otlp = { version = "0.31.1", default-features = false, features = [
+    "trace",
+    "grpc-tonic",
+] }
+opentelemetry_sdk = { version = "0.31.0", default-features = false, features = [
+    "trace",
+    "rt-tokio",
+] }
 oxrdf = "0.3"
 parking_lot = { version = "0.12.5", features = ["deadlock_detection"] }
 postcard = { version = "1.1.3", features = ["alloc"] }
@@ -110,7 +116,7 @@ thiserror = "2.0.18"
 tokio = { version = "1.52.1", features = ["full", "tracing"] }
 tokio-util = "0.7.18"
 tracing = "0.1.44"
-tracing-opentelemetry = "0.32.1"
+tracing-opentelemetry = { version = "0.32.1", default-features = false }
 tracing-subscriber = { version = "0.3.23", features = [
     "env-filter",
     "time",

api/src/auth.rs

@@ -2,7 +2,9 @@ use crate::error::{OidcError, ServerError, ServerResult, TokenError};
 use crate::server_state::ServerState;
 use crate::telemetry::record_auth_context;
 use aruna_core::errors::ConversionError;
-use aruna_core::structs::{AuthContext, OidcProviderConfig, Permission, RealmId, TokenClaims};
+use aruna_core::structs::{
+    AuthContext, OidcProviderConfig, Permission, RealmId, TokenClaims, blob_object_permission_path,
+};
 use aruna_operations::check_permissions::{CheckPermissionsConfig, CheckPermissionsOperation};
 use aruna_operations::driver::drive;
 use axum::extract::Request;
@@ -484,16 +486,21 @@ pub(crate) fn bucket_blob_permission_path(
     bucket: &str,
     key: &str,
 ) -> String {
-    format!(
-        "/{}/g/{group_id}/data/{}/{bucket}/{key}",
+    blob_object_permission_path(
         state.get_realm_id(),
+        group_id,
         state.get_node_id(),
+        bucket,
+        key,
     )
 }
 
 #[cfg(test)]
 mod test {
-    use crate::auth::{OIDC_PROVIDER_METADATA_CACHE_TTL_SECS, OidcValidator, extract_auth_context};
+    use crate::auth::{
+        OIDC_PROVIDER_METADATA_CACHE_TTL_SECS, OidcValidator, bucket_blob_permission_path,
+        extract_auth_context,
+    };
     use crate::server::ServerState;
     use aruna_core::UserId;
     use aruna_core::effects::{Effect, StorageEffect};
@@ -531,6 +538,47 @@ mod test {
     use tokio::sync::RwLock;
     use ulid::Ulid;
 
+    #[tokio::test]
+    async fn bucket_blob_permission_path_matches_canonical_blob_object_path() {
+        let storage_dir = tempfile::tempdir().unwrap();
+        let storage_handle =
+            storage::FjallStorage::open(storage_dir.path().to_str().unwrap()).unwrap();
+        let realm_id = RealmId::from_bytes(
+            *ed25519_dalek::SigningKey::from_bytes(&[7u8; 32])
+                .verifying_key()
+                .as_bytes(),
+        );
+        let node_id = iroh::SecretKey::generate().public();
+        let state = ServerState::new(
+            Arc::new(DriverContext {
+                storage_handle,
+                net_handle: None,
+                blob_handle: None,
+                automerge_handle: None,
+                metadata_handle: None,
+                task_handle: None,
+            }),
+            realm_id,
+            node_id,
+            NodeCapabilities::local_node(realm_id).unwrap(),
+            false,
+            None,
+        )
+        .await;
+
+        let group_id = Ulid::from_bytes([9u8; 16]);
+        assert_eq!(
+            bucket_blob_permission_path(&state, group_id, "bucket", "nested/file.txt"),
+            aruna_core::structs::blob_object_permission_path(
+                realm_id,
+                group_id,
+                node_id,
+                "bucket",
+                "nested/file.txt",
+            )
+        );
+    }
+
     #[derive(Clone)]
     struct OidcTestServerState {
         issuer: String,

api/src/routes/blobs.rs

@@ -1,7 +1,9 @@
 use crate::error::{ErrorResponse, ServerError, ServerResult};
 use crate::server_state::ServerState;
 use aruna_core::NodeId;
-use aruna_core::structs::{AuthContext, Permission};
+use aruna_core::structs::{
+    AuthContext, Permission, blob_bucket_permission_path, blob_object_permission_path,
+};
 use aruna_operations::check_permissions::{CheckPermissionsConfig, CheckPermissionsOperation};
 use aruna_operations::driver::drive;
 use aruna_operations::replication::protocol::ReplicationMode;
@@ -16,7 +18,7 @@ use axum::{Extension, Json, Router};
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;
 use std::sync::Arc;
-use tracing::{Instrument, info, warn};
+use tracing::{Instrument, debug, info, warn};
 use utoipa::{OpenApi, ToSchema};
 
 #[derive(OpenApi)]
@@ -87,17 +89,21 @@ pub async fn replicate_blob(
         Ok(None) => return Err(ServerError::NotFound),
     };
 
-    let mut permission_path = format!(
-        "/{}/g/{}/data/{}/{}",
-        state.get_realm_id(),
-        bucket_info.group_id,
-        state.get_node_id(),
-        request.bucket
-    );
-    if let Some(path) = request.path.as_deref() {
-        permission_path.push('/');
-        permission_path.push_str(path);
-    }
+    let permission_path = match request.path.as_deref() {
+        Some(path) => blob_object_permission_path(
+            state.get_realm_id(),
+            bucket_info.group_id,
+            state.get_node_id(),
+            &request.bucket,
+            path,
+        ),
+        None => blob_bucket_permission_path(
+            state.get_realm_id(),
+            bucket_info.group_id,
+            state.get_node_id(),
+            &request.bucket,
+        ),
+    };
 
     let allowed = drive(
         CheckPermissionsOperation::new(CheckPermissionsConfig {
@@ -166,6 +172,13 @@ pub async fn replicate_blob(
 
     tokio::spawn(
         async move {
+            debug!(
+                bucket = %bucket,
+                path = ?path,
+                version_id = ?version_id,
+                target_node = %target_node_id,
+                "Starting on-demand replication task"
+            );
             match drive(ReplicateScopeOperation::new(input), &ctx).await {
                 Ok(Some(Ok(result))) if result.failed == 0 => {
                     info!(bucket,