http shard client (#7)

* import cas_types definitions

* http based shard client

* use cas_types::Key

* add test

* fix linting

.gitignore

@@ -16,3 +16,4 @@ debug/
 # VS Code configs
 .vscode
 venv
+**/*.env

Cargo.toml

@@ -6,6 +6,8 @@ resolver = "2"
 members = [
     "cache",
     "cas_client",
+    "cas_object",
+    "cas_types",
     "data",
     "error_printer",
     "file_utils",
@@ -17,8 +19,6 @@ members = [
     "shard_client",
     "utils",
     "xet_error",
-    "cas_object",
-    "cas_types",
 ]
 
 exclude = [

data/src/remote_shard_interface.rs

@@ -335,8 +335,6 @@ impl RemoteShardInterface {
         }
 
         let salt = self.repo_salt()?;
-        let cas = self.cas()?;
-        let cas_ref = &cas;
         let shard_client = self.shard_client()?;
         let shard_client_ref = &shard_client;
         let shard_prefix = self.shard_prefix.clone();
@@ -352,25 +350,10 @@ impl RemoteShardInterface {
                 &si.shard_hash
             );
             let data = std::fs::read(&si.path)?;
-            let data_len = data.len();
-            // Upload the shard.
-            cas_ref
-                .put_bypass_stage(
-                    shard_prefix_ref,
-                    &si.shard_hash,
-                    data,
-                    vec![data_len as u64],
-                )
-                .await?;
 
-            debug!(
-                "Registering shard {shard_prefix_ref}/{:?} with shard server.",
-                &si.shard_hash
-            );
-
-            // That succeeded if we made it here, so now try to sync things.
+            // Upload the shard.
             shard_client_ref
-                .register_shard_with_salt(shard_prefix_ref, &si.shard_hash, false, &salt)
+                .upload_shard(&self.shard_prefix, &si.shard_hash, false, &data, &salt)
                 .await?;
 
             info!(
@@ -388,8 +371,6 @@ impl RemoteShardInterface {
             parutils::ParallelError::TaskError(e) => e,
         })?;
 
-        cas.flush().await?;
-
         Ok(())
     }
 

data/src/shard_interface.rs

@@ -1,8 +1,7 @@
 use super::configurations::{Endpoint::*, StorageConfig};
 use super::errors::Result;
-use crate::constants::XET_VERSION;
 use mdb_shard::ShardFileManager;
-use shard_client::{GrpcShardClient, LocalShardClient, ShardClientInterface};
+use shard_client::{HttpShardClient, LocalShardClient, ShardClientInterface};
 use std::sync::Arc;
 use tracing::{info, warn};
 
@@ -40,14 +39,7 @@ pub async fn create_shard_client(
 ) -> Result<Arc<dyn ShardClientInterface>> {
     info!("Shard endpoint = {:?}", shard_storage_config.endpoint);
     let client: Arc<dyn ShardClientInterface> = match &shard_storage_config.endpoint {
-        Server(endpoint) => {
-            let shard_connection_config = shard_client::ShardConnectionConfig {
-                endpoint: endpoint.clone(),
-                user_id: shard_storage_config.auth.user_id.clone(),
-                git_xet_version: XET_VERSION.to_string(),
-            };
-            Arc::new(GrpcShardClient::from_config(shard_connection_config).await?)
-        }
+        Server(endpoint) => Arc::new(HttpShardClient::new(endpoint)),
         FileSystem(path) => Arc::new(LocalShardClient::new(path).await?),
     };