fix(grpc): Load dependency proto files through reflection when availa…

…ble (tailcallhq#2683)

Co-authored-by: Kiryl Mialeshka <[email protected]>
Co-authored-by: Tushar Mathur <[email protected]>

src/core/proto_reader/fetch.rs

@@ -110,7 +110,14 @@ impl GrpcReflection {
             .execute(json!({"file_containing_symbol": service}))
             .await?;
 
-        request_proto(resp).await
+        request_proto(resp)
+    }
+
+    /// Makes `Get File` request to grpc reflection server
+    pub async fn get_file(&self, file_path: &str) -> Result<FileDescriptorProto> {
+        let resp = self.execute(json!({"file_by_filename": file_path})).await?;
+
+        request_proto(resp)
     }
 
     async fn execute(&self, body: serde_json::Value) -> Result<ReflectionResponse> {
@@ -158,7 +165,7 @@ impl GrpcReflection {
 }
 
 /// For extracting `FileDescriptorProto` from `CustomResponse`
-async fn request_proto(response: ReflectionResponse) -> Result<FileDescriptorProto> {
+fn request_proto(response: ReflectionResponse) -> Result<FileDescriptorProto> {
     let file_descriptor_resp = response
         .file_descriptor_response
         .context("Expected fileDescriptorResponse but found none")?;
@@ -196,6 +203,16 @@ mod grpc_fetch {
         BASE64_STANDARD.decode(bytes).unwrap()
     }
 
+    fn get_dto_file_descriptor() -> Vec<u8> {
+        let mut path = PathBuf::from(file!());
+        path.pop();
+        path.push("fixtures/dto_b64.txt");
+
+        let bytes = std::fs::read(path).unwrap();
+
+        BASE64_STANDARD.decode(bytes).unwrap()
+    }
+
     fn start_mock_server() -> httpmock::MockServer {
         httpmock::MockServer::start()
     }
@@ -228,6 +245,37 @@ mod grpc_fetch {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_dto_file() -> Result<()> {
+        let server = start_mock_server();
+
+        let http_reflection_file_mock = server.mock(|when, then| {
+            when.method(httpmock::Method::POST)
+                .path("/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo")
+                .body("\0\0\0\0\u{10}\u{1a}\u{0e}news_dto.proto");
+            then.status(200).body(get_dto_file_descriptor());
+        });
+
+        let grpc_reflection = GrpcReflection::new(
+            format!("http://localhost:{}", server.port()),
+            crate::core::runtime::test::init(None),
+        );
+
+        let runtime = crate::core::runtime::test::init(None);
+        let resp = grpc_reflection.get_file("news_dto.proto").await?;
+
+        let content = runtime
+            .file
+            .read(tailcall_fixtures::protobuf::NEWS_DTO)
+            .await?;
+        let expected = protox_parse::parse("news_dto.proto", &content)?;
+
+        assert_eq!(expected.name(), resp.name());
+
+        http_reflection_file_mock.assert();
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_resp_list_all() -> Result<()> {
         let server = start_mock_server();

src/core/proto_reader/fixtures/descriptor_b64.txt

@@ -1 +1 @@
-AAAABVgSEiIQbmV3cy5OZXdzU2VydmljZSLBCgq+CgoKbmV3cy5wcm90bxIEbmV3cxobZ29vZ2xlL3Byb3RvYnVmL2VtcHR5LnByb3RvIkIKBE5ld3MSCgoCaWQYASABKAUSDQoFdGl0bGUYAiABKAkSDAoEYm9keRgDIAEoCRIRCglwb3N0SW1hZ2UYBCABKAkiFAoGTmV3c0lkEgoKAmlkGAEgASgFIiMKDk11bHRpcGxlTmV3c0lkEhEKA2lkcxgBIAMyBk5ld3NJZCIcCghOZXdzTGlzdBIQCgRuZXdzGAEgAzIETmV3czLeAQoLTmV3c1NlcnZpY2USLQoKR2V0QWxsTmV3cxIVZ29vZ2xlLnByb3RvYnVmLkVtcHR5GghOZXdzTGlzdBIXCgdHZXROZXdzEgZOZXdzSWQaBE5ld3MSKwoPR2V0TXVsdGlwbGVOZXdzEg5NdWx0aXBsZU5ld3NJZBoITmV3c0xpc3QSKwoKRGVsZXRlTmV3cxIGTmV3c0lkGhVnb29nbGUucHJvdG9idWYuRW1wdHkSFgoIRWRpdE5ld3MSBE5ld3MaBE5ld3MSFQoHQWRkTmV3cxIETmV3cxoETmV3c0qGBwoGEgQAACABCggKAQISAwQADQoJCgIDABIDAgAlCgoKAgQAEgQGAAsBCgoKAwQAARIDBggMCgsKBAQAAgASAwcEEQoMCgUEAAIAARIDBwoMCgwKBQQAAgADEgMHDxAKDAoFBAACAAUSAwcECQoLCgQEAAIBEgMIBBUKDAoFBAACAQESAwgLEAoMCgUEAAIBAxIDCBMUCgwKBQQAAgEFEgMIBAoKCwoEBAACAhIDCQQUCgwKBQQAAgIBEgMJCw8KDAoFBAACAgMSAwkSEwoMCgUEAAICBRIDCQQKCgsKBAQAAgMSAwoEGQoMCgUEAAIDARIDCgsUCgwKBQQAAgMDEgMKFxgKDAoFBAACAwUSAwoECgoKCgIEARIEFgAYAQoKCgMEAQESAxYIDgoLCgQEAQIAEgMXBBEKDAoFBAECAAESAxcKDAoMCgUEAQIAAxIDFw8QCgwKBQQBAgAFEgMXBAkKCgoCBAISBBoAHAEKCgoDBAIBEgMaCBYKCwoEBAICABIDGwQcCgwKBQQCAgABEgMbFBcKDAoFBAICAAMSAxsaGwoMCgUEAgIABBIDGwQMCgwKBQQCAgAGEgMbDRMKCgoCBAMSBB4AIAEKCgoDBAMBEgMeCBAKCwoEBAMCABIDHwMaCgwKBQQDAgABEgMfERUKDAoFBAMCAAMSAx8YGQoMCgUEAwIABBIDHwMLCgwKBQQDAgAGEgMfDBAKCgoCBgASBA0AFAEKCgoDBgABEgMNCBMKCwoEBgACABIDDgRACgwKBQYAAgABEgMOCBIKDAoFBgACAAISAw4UKQoMCgUGAAIAAxIDDjQ8CgsKBAYAAgESAw8EKgoMCgUGAAIBARIDDwgPCgwKBQYAAgECEgMPERcKDAoFBgACAQMSAw8iJgoLCgQGAAICEgMQBD4KDAoFBgACAgESAxAIFwoMCgUGAAICAhIDEBknCgwKBQYAAgIDEgMQMjoKCwoEBgACAxIDEQQ+CgwKBQYAAgMBEgMRCBIKDAoFBgACAwISAxEUGgoMCgUGAAIDAxIDESU6CgsKBAYAAgQSAxIEKQoMCgUGAAIEARIDEggQCgwKBQYAAgQCEgMSEhYKDAoFBgACBAMSAxIhJQoLCgQGAAIFEgMTBCgKDAoFBgACBQESAxMIDwoMCgUGAAIFAhIDExEVCgwKBQYAAgUDEgMTICQKCAoBDBIDAAASYgZwcm90bzM=
+AAAAAyMSEiIQbmV3cy5OZXdzU2VydmljZSKMBgqJBgoKbmV3cy5wcm90bxIEbmV3cxoObmV3c19kdG8ucHJvdG8aG2dvb2dsZS9wcm90b2J1Zi9lbXB0eS5wcm90bzKoAgoLTmV3c1NlcnZpY2USNgoKR2V0QWxsTmV3cxIWLmdvb2dsZS5wcm90b2J1Zi5FbXB0eRoOLm5ld3MuTmV3c0xpc3QiABIlCgdHZXROZXdzEgwubmV3cy5OZXdzSWQaCi5uZXdzLk5ld3MiABI5Cg9HZXRNdWx0aXBsZU5ld3MSFC5uZXdzLk11bHRpcGxlTmV3c0lkGg4ubmV3cy5OZXdzTGlzdCIAEjQKCkRlbGV0ZU5ld3MSDC5uZXdzLk5ld3NJZBoWLmdvb2dsZS5wcm90b2J1Zi5FbXB0eSIAEiQKCEVkaXROZXdzEgoubmV3cy5OZXdzGgoubmV3cy5OZXdzIgASIwoHQWRkTmV3cxIKLm5ld3MuTmV3cxoKLm5ld3MuTmV3cyIASpQDCgYSBAAADgEKCAoBDBIDAAASCggKAQISAwIADQoJCgIDABIDBAAYCgkKAgMBEgMFACUKCgoCBgASBAcADgEKCgoDBgABEgMHCBMKCwoEBgACABIDCAI9CgwKBQYAAgABEgMIBhAKDAoFBgACAAISAwgRJgoMCgUGAAIAAxIDCDE5CgsKBAYAAgESAwkCJwoMCgUGAAIBARIDCQYNCgwKBQYAAgECEgMJDhQKDAoFBgACAQMSAwkfIwoLCgQGAAICEgMKAjsKDAoFBgACAgESAwoGFQoMCgUGAAICAhIDChYkCgwKBQYAAgIDEgMKLzcKCwoEBgACAxIDCwI7CgwKBQYAAgMBEgMLBhAKDAoFBgACAwISAwsRFwoMCgUGAAIDAxIDCyI3CgsKBAYAAgQSAwwCJgoMCgUGAAIEARIDDAYOCgwKBQYAAgQCEgMMDxMKDAoFBgACBAMSAwweIgoLCgQGAAIFEgMNAiUKDAoFBgACBQESAw0GDQoMCgUGAAIFAhIDDQ4SCgwKBQYAAgUDEgMNHSFiBnByb3RvMw==

src/core/proto_reader/fixtures/dto_b64.txt

@@ -0,0 +1 @@
+AAAABHESEBoObmV3c19kdG8ucHJvdG8i3AgK2QgKDm5ld3NfZHRvLnByb3RvEgRuZXdzGhtnb29nbGUvcHJvdG9idWYvZW1wdHkucHJvdG8ihAEKBE5ld3MSDgoCaWQYASABKAVSAmlkEhQKBXRpdGxlGAIgASgJUgV0aXRsZRISCgRib2R5GAMgASgJUgRib2R5EhwKCXBvc3RJbWFnZRgEIAEoCVIJcG9zdEltYWdlEiQKBnN0YXR1cxgFIAEoDjIMLm5ld3MuU3RhdHVzUgZzdGF0dXMiGAoGTmV3c0lkEg4KAmlkGAEgASgFUgJpZCIwCg5NdWx0aXBsZU5ld3NJZBIeCgNpZHMYASADKAsyDC5uZXdzLk5ld3NJZFIDaWRzIioKCE5ld3NMaXN0Eh4KBG5ld3MYASADKAsyCi5uZXdzLk5ld3NSBG5ld3MqLwoGU3RhdHVzEg0KCVBVQkxJU0hFRBAAEgkKBURSQUZUEAESCwoHREVMRVRFRBACSusFCgYSBAAAGCwKCAoBDBIDAAASCggKAQISAwIADQoJCgIDABIDBAAlCgoKAgUAEgQGAAoBCgoKAwUAARIDBgULCgsKBAUAAgASAwcCEAoMCgUFAAIAARIDBwILCgwKBQUAAgACEgMHDg8KCwoEBQACARIDCAIMCgwKBQUAAgEBEgMIAgcKDAoFBQACAQISAwgKCwoLCgQFAAICEgMJAg4KDAoFBQACAgESAwkCCQoMCgUFAAICAhIDCQwNCgoKAgQAEgQMABIBCgoKAwQAARIDDAgMCgsKBAQAAgASAw0CDwoMCgUEAAIABRIDDQIHCgwKBQQAAgABEgMNCAoKDAoFBAACAAMSAw0NDgoLCgQEAAIBEgMOAhMKDAoFBAACAQUSAw4CCAoMCgUEAAIBARIDDgkOCgwKBQQAAgEDEgMOERIKCwoEBAACAhIDDwISCgwKBQQAAgIFEgMPAggKDAoFBAACAgESAw8JDQoMCgUEAAICAxIDDxARCgsKBAQAAgMSAxACFwoMCgUEAAIDBRIDEAIICgwKBQQAAgMBEgMQCRIKDAoFBAACAwMSAxAVFgoLCgQEAAIEEgMRAhQKDAoFBAACBAYSAxECCAoMCgUEAAIEARIDEQkPCgwKBQQAAgQDEgMREhMKCQoCBAESAxQAIAoKCgMEAQESAxQIDgoLCgQEAQIAEgMUER4KDAoFBAECAAUSAxQRFgoMCgUEAQIAARIDFBcZCgwKBQQBAgADEgMUHB0KCQoCBAISAxYAMwoKCgMEAgESAxYIFgoLCgQEAgIAEgMWGTEKDAoFBAICAAQSAxYZIQoMCgUEAgIABhIDFiIoCgwKBQQCAgABEgMWKSwKDAoFBAICAAMSAxYvMAoJCgIEAxIDGAAsCgoKAwQDARIDGAgQCgsKBAQDAgASAxgTKgoMCgUEAwIABBIDGBMbCgwKBQQDAgAGEgMYHCAKDAoFBAMCAAESAxghJQoMCgUEAwIAAxIDGCgpYgZwcm90bzM=

src/core/proto_reader/reader.rs

@@ -1,15 +1,18 @@
 use std::collections::{HashMap, VecDeque};
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 
 use anyhow::Context;
-use futures_util::future::join_all;
+use futures_util::future::{join_all, BoxFuture};
+use futures_util::FutureExt;
 use prost_reflect::prost_types::{FileDescriptorProto, FileDescriptorSet};
 use protox::file::{FileResolver, GoogleFileResolver};
 
 use crate::core::proto_reader::fetch::GrpcReflection;
 use crate::core::resource_reader::{Cached, ResourceReader};
 use crate::core::runtime::TargetRuntime;
 
+#[derive(Clone)]
 pub struct ProtoReader {
     reader: ResourceReader<Cached>,
     runtime: TargetRuntime,
@@ -29,7 +32,7 @@ impl ProtoReader {
 
     /// Fetches proto files from a grpc server (grpc reflection)
     pub async fn fetch<T: AsRef<str>>(&self, url: T) -> anyhow::Result<Vec<ProtoMetadata>> {
-        let grpc_reflection = GrpcReflection::new(url.as_ref(), self.runtime.clone());
+        let grpc_reflection = Arc::new(GrpcReflection::new(url.as_ref(), self.runtime.clone()));
 
         let mut proto_metadata = vec![];
         let service_list = grpc_reflection.list_all_files().await?;
@@ -39,7 +42,9 @@ impl ProtoReader {
             }
             let file_descriptor_proto = grpc_reflection.get_by_service(&service).await?;
             Self::check_package(&file_descriptor_proto)?;
-            let descriptors = self.resolve(file_descriptor_proto, None).await?;
+            let descriptors = self
+                .reflection_resolve(grpc_reflection.clone(), file_descriptor_proto)
+                .await?;
             let metadata = ProtoMetadata {
                 descriptor_set: FileDescriptorSet { file: descriptors },
                 path: url.as_ref().to_string(),
@@ -64,7 +69,7 @@ impl ProtoReader {
         Self::check_package(&file_read)?;
 
         let descriptors = self
-            .resolve(file_read, PathBuf::from(path.as_ref()).parent())
+            .file_resolve(file_read, PathBuf::from(path.as_ref()).parent())
             .await?;
         let metadata = ProtoMetadata {
             descriptor_set: FileDescriptorSet { file: descriptors },
@@ -73,12 +78,15 @@ impl ProtoReader {
         Ok(metadata)
     }
 
-    /// Performs BFS to import all nested proto files
-    async fn resolve(
+    /// Used as a helper file to resolve dependencies proto files
+    async fn resolve_dependencies<F>(
         &self,
         parent_proto: FileDescriptorProto,
-        parent_path: Option<&Path>,
-    ) -> anyhow::Result<Vec<FileDescriptorProto>> {
+        resolve_fn: F,
+    ) -> anyhow::Result<Vec<FileDescriptorProto>>
+    where
+        F: Fn(&str) -> BoxFuture<'_, anyhow::Result<FileDescriptorProto>>,
+    {
         let mut descriptors: HashMap<String, FileDescriptorProto> = HashMap::new();
         let mut queue = VecDeque::new();
         queue.push_back(parent_proto.clone());
@@ -87,7 +95,7 @@ impl ProtoReader {
             let futures: Vec<_> = file
                 .dependency
                 .iter()
-                .map(|import| self.read_proto(import, parent_path))
+                .map(|import| resolve_fn(import))
                 .collect();
 
             let results = join_all(futures).await;
@@ -108,6 +116,34 @@ impl ProtoReader {
         Ok(descriptors_vec)
     }
 
+    /// Used to resolve dependencies proto files using file reader
+    async fn file_resolve(
+        &self,
+        parent_proto: FileDescriptorProto,
+        parent_path: Option<&Path>,
+    ) -> anyhow::Result<Vec<FileDescriptorProto>> {
+        self.resolve_dependencies(parent_proto, |import| {
+            let parent_path = parent_path.map(|p| p.to_path_buf());
+            let this = self.clone();
+
+            async move { this.read_proto(import, parent_path.as_deref()).await }.boxed()
+        })
+        .await
+    }
+
+    /// Used to resolve dependencies proto files using reflection
+    async fn reflection_resolve(
+        &self,
+        grpc_reflection: Arc<GrpcReflection>,
+        parent_proto: FileDescriptorProto,
+    ) -> anyhow::Result<Vec<FileDescriptorProto>> {
+        self.resolve_dependencies(parent_proto, |file| {
+            let grpc_reflection = Arc::clone(&grpc_reflection);
+            async move { grpc_reflection.get_file(file).await }.boxed()
+        })
+        .await
+    }
+
     /// Tries to load well-known google proto files and if not found uses normal
     /// file and http IO to resolve them
     async fn read_proto<T: AsRef<str>>(
@@ -180,7 +216,7 @@ mod test_proto_config {
 
         let reader = ProtoReader::init(ResourceReader::<Cached>::cached(runtime.clone()), runtime);
         let file_descriptors = reader
-            .resolve(reader.read_proto(&test_file, None).await?, Some(test_dir))
+            .file_resolve(reader.read_proto(&test_file, None).await?, Some(test_dir))
             .await?;
         for file in file_descriptors
             .iter()

tailcall-cloudflare/tests/cf_tests.spec.ts

@@ -8,10 +8,12 @@ describe("fetch", () => {
     let placeholder_batch = (await readFile("../examples/jsonplaceholder_batch.graphql")).toString()
     let grpc = (await readFile("../examples/grpc.graphql")).toString()
     let news_proto = (await readFile("../tailcall-fixtures/fixtures/protobuf/news.proto")).toString()
+    let news_dto_proto = (await readFile("../tailcall-fixtures/fixtures/protobuf/news_dto.proto")).toString()
 
     let bucket = await mf.getR2Bucket("MY_R2")
     await bucket.put("examples/grpc.graphql", grpc)
     await bucket.put("examples/../tailcall-fixtures/fixtures/protobuf/news.proto", news_proto)
+    await bucket.put("examples/../tailcall-fixtures/fixtures/protobuf/news_dto.proto", news_dto_proto)
     await bucket.put("tailcall-fixtures/fixtures/protobuf/news.proto", grpc)
     await bucket.put("examples/jsonplaceholder.graphql", placeholder)
     await bucket.put("examples/jsonplaceholder_batch.graphql", placeholder_batch)

tailcall-fixtures/README.md

@@ -1,3 +1,12 @@
 # Tailcall fixtures
 
 Package that contains configs and fixtures used in tests and examples to be shared among different parts of tailcall plus helper functions to resolve these files in tests.
+
+## gRPC binary files
+
+Instructions to generate those files:
+
+1. go to `src/core/proto_reader/fetch.rs`
+2. modify `GrpcReflection.execute` function to print the request `body` as string and response `body` as base64 encoded
+3. convert the base64 into bin files using online base64 to file websites or manually
+4. rename the files and replace the old ones

tailcall-fixtures/fixtures/protobuf/news.proto

@@ -1,22 +1,9 @@
 syntax = "proto3";
 
-import "google/protobuf/empty.proto";
-
 package news;
 
-enum Status {
-  PUBLISHED = 0;
-  DRAFT = 1;
-  DELETED = 2;
-}
-
-message News {
-  int32 id = 1;
-  string title = 2;
-  string body = 3;
-  string postImage = 4;
-  Status status = 5;
-}
+import "news_dto.proto";
+import "google/protobuf/empty.proto";
 
 service NewsService {
   rpc GetAllNews(google.protobuf.Empty) returns (NewsList) {}
@@ -26,9 +13,3 @@ service NewsService {
   rpc EditNews(News) returns (News) {}
   rpc AddNews(News) returns (News) {}
 }
-
-message NewsId { int32 id = 1; }
-
-message MultipleNewsId { repeated NewsId ids = 1; }
-
-message NewsList { repeated News news = 1; }

tailcall-fixtures/fixtures/protobuf/news_dto.proto

@@ -0,0 +1,25 @@
+syntax = "proto3";
+
+package news;
+
+import "google/protobuf/empty.proto";
+
+enum Status {
+  PUBLISHED = 0;
+  DRAFT = 1;
+  DELETED = 2;
+}
+
+message News {
+  int32 id = 1;
+  string title = 2;
+  string body = 3;
+  string postImage = 4;
+  Status status = 5;
+}
+
+message NewsId { int32 id = 1; }
+
+message MultipleNewsId { repeated NewsId ids = 1; }
+
+message NewsList { repeated News news = 1; }

tests/core/snapshots/grpc-reflection.md_client.snap

@@ -28,6 +28,7 @@ type News {
   body: String
   id: Int
   postImage: String
+  status: Status
   title: String
 }
 
@@ -41,6 +42,12 @@ type Query {
   news: NewsData!
 }
 
+enum Status {
+  DELETED
+  DRAFT
+  PUBLISHED
+}
+
 scalar UInt128
 
 scalar UInt16

tests/core/snapshots/grpc-reflection.md_merged.snap

@@ -9,10 +9,17 @@ schema
   query: Query
 }
 
+enum Status {
+  DELETED
+  DRAFT
+  PUBLISHED
+}
+
 type News {
   body: String
   id: Int
   postImage: String
+  status: Status
   title: String
 }
 

tests/execution/grpc-reflection.md

@@ -21,6 +21,13 @@ type News {
   title: String
   body: String
   postImage: String
+  status: Status
+}
+
+enum Status {
+  PUBLISHED
+  DRAFT
+  DELETED
 }
 ```
 
@@ -31,15 +38,32 @@ type News {
     textBody: \0\0\0\0\x02:\0
   response:
     status: 200
-    fileBody: grpc/reflection/news-list-services.bin
+    fileBody: grpc/reflection/list-services.bin
 
 - request:
     method: POST
     url: http://localhost:50051/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo
     textBody: \0\0\0\0\x12\"\x10news.NewsService
   response:
     status: 200
-    fileBody: grpc/reflection/news-service-descriptor.bin
+    fileBody: grpc/reflection/news-service.bin
+
+- request:
+    method: POST
+    url: http://localhost:50051/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo
+    textBody: \0\0\0\0\x1d\x1a\x1bgoogle/protobuf/empty.proto
+  expectedHits: 2
+  response:
+    status: 200
+    fileBody: grpc/reflection/protobuf_empty.bin
+
+- request:
+    method: POST
+    url: http://localhost:50051/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo
+    textBody: \0\0\0\0\x10\x1a\x0enews_dto.proto
+  response:
+    status: 200
+    fileBody: grpc/reflection/news_dto.bin
 
 - request:
     method: POST