Task rdma

Cargo.toml

@@ -30,6 +30,8 @@ members = [
     "lib/crowdb-access-iceberg",
     "lib/crowdb-access-dataset",
     "lib/crowdb-access-multipart",
+    "lib/crowdb-sdk-native",
+    "lib/crowdb-sdk-native-python",
 ]
 exclude = ["third-party/hyper", "sdk"]
 # : crowdb-tree/ffi moved from `exclude` into `members` now that

app/crowdb-access-server/Cargo.toml

@@ -55,6 +55,7 @@ thiserror = { workspace = true }
 sync_wrapper = { version = "1", features = ["futures"] }
 
 [dev-dependencies]
+crowdb-access-dataset = { path = "../../lib/crowdb-access-dataset", features = ["test-util"] }
 crowdb-kv = { path = "../../lib/crowdb-kv" }
 bytes = "1"
 tempfile = "3"

app/crowdb-access-server/src/config.rs

@@ -58,6 +58,7 @@ pub struct CommonConfig {
     pub management_seeds: Vec<String>,
     pub diskio_connections_per_endpoint: usize,
     pub diskio_rpc_workers: u32,
+    pub chunk_kv_rpc: crowdb_chunk_kv_client::ChunkKvRpcTransportConfig,
 }
 
 impl Default for CommonConfig {
@@ -66,6 +67,7 @@ impl Default for CommonConfig {
             management_seeds: Vec::new(),
             diskio_connections_per_endpoint: 2,
             diskio_rpc_workers: 2,
+            chunk_kv_rpc: crowdb_chunk_kv_client::ChunkKvRpcTransportConfig::default(),
         }
     }
 }
@@ -249,7 +251,6 @@ pub struct DatasetConfig {
     pub max_in_flight: usize,
     pub max_batch_size: usize,
     pub request_timeout_seconds: u64,
-    pub lease_ttl_seconds: u64,
 }
 
 impl Default for DatasetConfig {
@@ -259,7 +260,6 @@ impl Default for DatasetConfig {
             max_in_flight: 8,
             max_batch_size: 10_000,
             request_timeout_seconds: 60,
-            lease_ttl_seconds: 60,
         }
     }
 }
@@ -311,6 +311,10 @@ pub struct IcebergGcConfig {
 impl BaseConfig for AccessConfig {
     #[allow(clippy::too_many_lines)]
     fn validate(&self) -> Result<(), String> {
+        self.common
+            .chunk_kv_rpc
+            .validate()
+            .map_err(|error| error.to_string())?;
         if self.common.diskio_connections_per_endpoint == 0 || self.common.diskio_rpc_workers == 0 {
             return Err("common DiskIO connections and RPC workers must be nonzero".into());
         }
@@ -432,9 +436,6 @@ impl AccessConfig {
         if self.dataset.request_timeout_seconds == 0 || self.dataset.request_timeout_seconds > 300 {
             return Err("dataset.request_timeout_seconds must be between 1 and 300".into());
         }
-        if self.dataset.lease_ttl_seconds == 0 || self.dataset.lease_ttl_seconds > 3600 {
-            return Err("dataset.lease_ttl_seconds must be between 1 and 3600".into());
-        }
         Ok(())
     }
 

app/crowdb-access-server/src/credentials.rs

@@ -34,8 +34,6 @@ pub enum CredentialAuthorityError {
     InvalidKey,
     #[error("S3 user has multiple credential records")]
     DuplicateUser,
-    #[error("S3 user credential preparation is malformed or belongs to another user")]
-    InvalidUserPreparation,
     #[error("S3 user credential is disabled or malformed")]
     InvalidUserCredential,
 }
@@ -116,11 +114,11 @@ impl CredentialAuthority {
             if found.is_some() {
                 return Err(CredentialAuthorityError::DuplicateUser);
             }
-            let credential = self.cipher.decrypt(user, access_key, &record.encrypted)?;
+            let mut credential = self.cipher.decrypt(user, access_key, &record.encrypted)?;
             if !credential.enabled {
                 return Err(CredentialAuthorityError::InvalidUserCredential);
             }
-            let secret_key = String::from_utf8(credential.secret_key.clone())
+            let secret_key = String::from_utf8(std::mem::take(&mut credential.secret_key))
                 .map_err(|_| CredentialAuthorityError::InvalidUserCredential)?;
             found = Some(IssuedUserToken {
                 access_key_id: access_key.to_owned(),