about summary refs log tree commit diff
path: root/tvix/store/src
diff options
context:
space:
mode:
authorFlorian Klink <flokli@flokli.de>2023-11-13T12·32+0200
committerflokli <flokli@flokli.de>2023-11-15T06·43+0000
commit30e0c320666f8ecaf37f6d966e45c40f988cce78 (patch)
tree808bdb2c8832e2083bc17493dc8982b755486824 /tvix/store/src
parentc83841d3a13a5d5b3fdaa7f9beecd743f1283691 (diff)
refactor(tvix/castore/tonic): make async, support wait-connect=? r/7022
This moves the sync `channel::from_url` to a async
`tonic::channel_from_url`. It now allows connecting non-lazily if `wait-
connect=1` is set in the URL params.

Also, make the pingpong tests for blobsvc and directorysvc use the wait-
connect=1 codepath.

Change-Id: Ibeea33117c8121814627e7f6aba0e943ae2e92ca
Reviewed-on: https://cl.tvl.fyi/c/depot/+/10030
Tested-by: BuildkiteCI
Reviewed-by: Connor Brewster <cbrewster@hey.com>
Diffstat (limited to 'tvix/store/src')
-rw-r--r--tvix/store/src/bin/tvix-store.rs28
-rw-r--r--tvix/store/src/pathinfoservice/from_addr.rs19
-rw-r--r--tvix/store/src/pathinfoservice/grpc.rs4
3 files changed, 26 insertions, 25 deletions
diff --git a/tvix/store/src/bin/tvix-store.rs b/tvix/store/src/bin/tvix-store.rs
index 3f7d984cd014..2669073c9140 100644
--- a/tvix/store/src/bin/tvix-store.rs
+++ b/tvix/store/src/bin/tvix-store.rs
@@ -193,13 +193,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             path_info_service_addr,
         } => {
             // initialize stores
-            let blob_service = blobservice::from_addr(&blob_service_addr)?;
-            let directory_service = directoryservice::from_addr(&directory_service_addr)?;
+            let blob_service = blobservice::from_addr(&blob_service_addr).await?;
+            let directory_service = directoryservice::from_addr(&directory_service_addr).await?;
             let path_info_service = pathinfoservice::from_addr(
                 &path_info_service_addr,
                 blob_service.clone(),
                 directory_service.clone(),
-            )?;
+            )
+            .await?;
 
             let listen_address = listen_address
                 .unwrap_or_else(|| "[::]:8000".to_string())
@@ -247,13 +248,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             path_info_service_addr,
         } => {
             // FUTUREWORK: allow flat for single files?
-            let blob_service = blobservice::from_addr(&blob_service_addr)?;
-            let directory_service = directoryservice::from_addr(&directory_service_addr)?;
+            let blob_service = blobservice::from_addr(&blob_service_addr).await?;
+            let directory_service = directoryservice::from_addr(&directory_service_addr).await?;
             let path_info_service = pathinfoservice::from_addr(
                 &path_info_service_addr,
                 blob_service.clone(),
                 directory_service.clone(),
-            )?;
+            )
+            .await?;
 
             let tasks = paths
                 .into_iter()
@@ -343,13 +345,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             list_root,
             threads,
         } => {
-            let blob_service = blobservice::from_addr(&blob_service_addr)?;
-            let directory_service = directoryservice::from_addr(&directory_service_addr)?;
+            let blob_service = blobservice::from_addr(&blob_service_addr).await?;
+            let directory_service = directoryservice::from_addr(&directory_service_addr).await?;
             let path_info_service = pathinfoservice::from_addr(
                 &path_info_service_addr,
                 blob_service.clone(),
                 directory_service.clone(),
-            )?;
+            )
+            .await?;
 
             let mut fuse_daemon = tokio::task::spawn_blocking(move || {
                 let f = TvixStoreFs::new(
@@ -383,13 +386,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             path_info_service_addr,
             list_root,
         } => {
-            let blob_service = blobservice::from_addr(&blob_service_addr)?;
-            let directory_service = directoryservice::from_addr(&directory_service_addr)?;
+            let blob_service = blobservice::from_addr(&blob_service_addr).await?;
+            let directory_service = directoryservice::from_addr(&directory_service_addr).await?;
             let path_info_service = pathinfoservice::from_addr(
                 &path_info_service_addr,
                 blob_service.clone(),
                 directory_service.clone(),
-            )?;
+            )
+            .await?;
 
             tokio::task::spawn_blocking(move || {
                 let fs = TvixStoreFs::new(
diff --git a/tvix/store/src/pathinfoservice/from_addr.rs b/tvix/store/src/pathinfoservice/from_addr.rs
index 01764b251be4..0f1f8d5c96a6 100644
--- a/tvix/store/src/pathinfoservice/from_addr.rs
+++ b/tvix/store/src/pathinfoservice/from_addr.rs
@@ -23,7 +23,7 @@ use url::Url;
 ///
 /// As the [PathInfoService] needs to talk to [BlobService] and [DirectoryService],
 /// these also need to be passed in.
-pub fn from_addr(
+pub async fn from_addr(
     uri: &str,
     blob_service: Arc<dyn BlobService>,
     directory_service: Arc<dyn DirectoryService>,
@@ -68,7 +68,7 @@ pub fn from_addr(
         // - In the case of unix sockets, there must be a path, but may not be a host.
         // - In the case of non-unix sockets, there must be a host, but no path.
         // Constructing the channel is handled by tvix_castore::channel::from_url.
-        let client = PathInfoServiceClient::new(tvix_castore::channel::from_url(&url)?);
+        let client = PathInfoServiceClient::new(tvix_castore::tonic::channel_from_url(&url).await?);
         Arc::new(GRPCPathInfoService::from_client(client))
     } else {
         Err(Error::StorageError(format!(
@@ -91,6 +91,8 @@ mod tests {
         static ref TMPDIR_SLED_2: TempDir = TempDir::new().unwrap();
     }
 
+    // the gRPC tests below don't fail, because we connect lazily.
+
     /// This uses a unsupported scheme.
     #[test_case("http://foo.example/test", false; "unsupported scheme")]
     /// This configures sled in temporary mode.
@@ -111,15 +113,6 @@ mod tests {
     #[test_case("memory:///", false; "memory invalid root path")]
     /// This sets a memory url path to "/foo", which is invalid.
     #[test_case("memory:///foo", false; "memory invalid root path foo")]
-    fn test_from_addr(uri_str: &str, is_ok: bool) {
-        assert_eq!(
-            from_addr(uri_str, gen_blob_service(), gen_directory_service()).is_ok(),
-            is_ok
-        )
-    }
-
-    // the gRPC tests below don't fail, because we connect lazily.
-
     /// Correct scheme to connect to a unix socket.
     #[test_case("grpc+unix:///path/to/somewhere", true; "grpc valid unix socket")]
     /// Correct scheme for unix socket, but setting a host too, which is invalid.
@@ -135,7 +128,9 @@ mod tests {
     #[tokio::test]
     async fn test_from_addr_tokio(uri_str: &str, is_ok: bool) {
         assert_eq!(
-            from_addr(uri_str, gen_blob_service(), gen_directory_service()).is_ok(),
+            from_addr(uri_str, gen_blob_service(), gen_directory_service())
+                .await
+                .is_ok(),
             is_ok
         )
     }
diff --git a/tvix/store/src/pathinfoservice/grpc.rs b/tvix/store/src/pathinfoservice/grpc.rs
index 2e58a464aa46..ef3b0b77ec54 100644
--- a/tvix/store/src/pathinfoservice/grpc.rs
+++ b/tvix/store/src/pathinfoservice/grpc.rs
@@ -180,7 +180,9 @@ mod tests {
             let url = url::Url::parse(&format!("grpc+unix://{}", socket_path.display()))
                 .expect("must parse");
             let client = PathInfoServiceClient::new(
-                tvix_castore::channel::from_url(&url).expect("must succeed"),
+                tvix_castore::tonic::channel_from_url(&url)
+                    .await
+                    .expect("must succeed"),
             );
 
             GRPCPathInfoService::from_client(client)