From bd2c6d4f4840ff4642563ee1ee5f9058a5dbee69 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Mon, 6 Apr 2026 11:04:10 +0300 Subject: [PATCH] feat: add grpc server streaming adapter --- Cargo.lock | 1 + TASKS.md | 10 +- crates/crank-adapter-grpc/Cargo.toml | 1 + crates/crank-adapter-grpc/proto/echo.proto | 1 + crates/crank-adapter-grpc/src/client.rs | 187 +++++++++++++++--- crates/crank-adapter-grpc/src/error.rs | 6 +- crates/crank-adapter-grpc/src/lib.rs | 2 +- crates/crank-adapter-grpc/src/model.rs | 17 ++ crates/crank-adapter-grpc/src/test_support.rs | 25 +++ crates/crank-runtime/src/executor.rs | 139 ++++++++++++- 10 files changed, 358 insertions(+), 31 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6ce1cb8..d690f90 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -363,6 +363,7 @@ dependencies = [ "base64", "crank-core", "crank-proto", + "futures-util", "prost", "prost-reflect", "prost-types", diff --git a/TASKS.md b/TASKS.md index dcc08c9..40447f8 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,16 +2,16 @@ ## Current -### `feat/rest-sse-adapter` +### `feat/grpc-server-streaming-adapter` Status: completed DoD: -- REST adapter supports bounded SSE collection for `window` mode -- SSE events are parsed into normalized JSON items +- gRPC adapter supports bounded server-stream collection for `window` mode +- streamed protobuf messages are decoded into normalized JSON items - adapter respects item/window limits and upstream timeout -- runtime dispatches REST `window` mode through SSE adapter path -- local SSE upstream tests cover success, timeout and malformed events +- runtime dispatches gRPC `window` mode through streaming adapter path +- local gRPC upstream tests cover success, timeout and malformed method kind handling ## Next diff --git a/crates/crank-adapter-grpc/Cargo.toml b/crates/crank-adapter-grpc/Cargo.toml index aafb201..182b2e1 100644 --- a/crates/crank-adapter-grpc/Cargo.toml +++ b/crates/crank-adapter-grpc/Cargo.toml @@ -12,6 +12,7 @@ test-support = [] base64.workspace = true crank-core = { path = "../crank-core" } crank-proto = { path = "../crank-proto" } +futures-util = "0.3" prost.workspace = true prost-reflect.workspace = true serde.workspace = true diff --git a/crates/crank-adapter-grpc/proto/echo.proto b/crates/crank-adapter-grpc/proto/echo.proto index c26a235..0ac81c0 100644 --- a/crates/crank-adapter-grpc/proto/echo.proto +++ b/crates/crank-adapter-grpc/proto/echo.proto @@ -4,6 +4,7 @@ package echo; service EchoService { rpc UnaryEcho(EchoRequest) returns (EchoResponse); + rpc ServerEcho(EchoRequest) returns (stream EchoResponse); } message EchoRequest { diff --git a/crates/crank-adapter-grpc/src/client.rs b/crates/crank-adapter-grpc/src/client.rs index b58e7db..024a399 100644 --- a/crates/crank-adapter-grpc/src/client.rs +++ b/crates/crank-adapter-grpc/src/client.rs @@ -2,8 +2,11 @@ use std::{collections::BTreeMap, str::FromStr, time::Duration}; use base64::{Engine as _, engine::general_purpose::STANDARD}; use crank_core::GrpcTarget; +use futures_util::StreamExt; use prost::Message; use prost_reflect::{DescriptorPool, MethodDescriptor, prost_types::FileDescriptorSet}; +use serde_json::json; +use tokio::time::{Instant, timeout_at}; use tonic::{ Request, client::Grpc, @@ -11,7 +14,10 @@ use tonic::{ transport::Endpoint, }; -use crate::{GrpcAdapterError, GrpcRequest, GrpcResponse, codec::JsonCodec}; +use crate::{ + GrpcAdapterError, GrpcRequest, GrpcResponse, GrpcWindowRequest, GrpcWindowResponse, + codec::JsonCodec, +}; #[derive(Clone, Debug, Default)] pub struct GrpcAdapter; @@ -34,26 +40,8 @@ impl GrpcAdapter { }); } - let endpoint = Endpoint::from_shared(target.server_addr.clone())? - .timeout(Duration::from_millis(request.timeout_ms)); - let channel = endpoint.connect().await?; - let mut grpc = Grpc::new(channel); - grpc.ready() - .await - .map_err(|error| GrpcAdapterError::Status { - code: tonic::Code::Unavailable, - message: error.to_string(), - })?; - - let service_name = method.parent_service().full_name().to_owned(); - let method_name = method.name().to_owned(); - let path = tonic::codegen::http::uri::PathAndQuery::from_str(&format!( - "/{service_name}/{method_name}" - )) - .map_err(|_| GrpcAdapterError::InvalidMethodPath { - service: service_name, - method: method_name, - })?; + let mut grpc = connect(target, request.timeout_ms).await?; + let path = build_path(&method)?; let codec = JsonCodec::new(method.input(), method.output()); let mut tonic_request = Request::new(request.body.clone()); @@ -69,6 +57,95 @@ impl GrpcAdapter { body, }) } + + pub async fn execute_window( + &self, + target: &GrpcTarget, + request: &GrpcWindowRequest, + ) -> Result { + let method = resolve_method(target)?; + if method.is_client_streaming() || !method.is_server_streaming() { + return Err(GrpcAdapterError::UnsupportedStreamingMethodKind { + service: format!("{}.{}", target.package, target.service), + method: target.method.clone(), + }); + } + + let mut grpc = connect(target, request.request.timeout_ms).await?; + let path = build_path(&method)?; + let codec = JsonCodec::new(method.input(), method.output()); + let mut tonic_request = Request::new(request.request.body.clone()); + + apply_headers(&mut tonic_request, &request.request.headers)?; + + let response = grpc.server_streaming(tonic_request, path, codec).await?; + let headers = normalize_headers(response.metadata()); + let mut stream = response.into_inner(); + let deadline = Instant::now() + Duration::from_millis(request.window_duration_ms); + let mut items = Vec::new(); + let mut done = true; + + loop { + if request + .max_items + .is_some_and(|limit| items.len() >= limit as usize) + { + done = false; + break; + } + + let next_message = match timeout_at(deadline, stream.next()).await { + Ok(next_message) => next_message, + Err(_) => break, + }; + + let Some(next_message) = next_message else { + break; + }; + + let message = next_message?; + items.push(message); + } + + Ok(GrpcWindowResponse { + status_code: 200, + headers, + body: json!({ + "items": items, + "done": done, + }), + }) + } +} + +async fn connect( + target: &GrpcTarget, + timeout_ms: u64, +) -> Result, GrpcAdapterError> { + let endpoint = Endpoint::from_shared(target.server_addr.clone())? + .timeout(Duration::from_millis(timeout_ms)); + let channel = endpoint.connect().await?; + let mut grpc = Grpc::new(channel); + grpc.ready() + .await + .map_err(|error| GrpcAdapterError::Status { + code: tonic::Code::Unavailable, + message: error.to_string(), + })?; + + Ok(grpc) +} + +fn build_path( + method: &MethodDescriptor, +) -> Result { + let service_name = method.parent_service().full_name().to_owned(); + let method_name = method.name().to_owned(); + tonic::codegen::http::uri::PathAndQuery::from_str(&format!("/{service_name}/{method_name}")) + .map_err(|_| GrpcAdapterError::InvalidMethodPath { + service: service_name, + method: method_name, + }) } fn resolve_method(target: &GrpcTarget) -> Result { @@ -141,7 +218,7 @@ mod tests { use crank_core::{DescriptorId, GrpcTarget}; use serde_json::json; - use crate::{GrpcAdapter, GrpcRequest, test_support}; + use crate::{GrpcAdapter, GrpcAdapterError, GrpcRequest, GrpcWindowRequest, test_support}; #[tokio::test] async fn executes_unary_grpc_request() { @@ -165,4 +242,70 @@ mod tests { assert_eq!(response.body, json!({ "message": "hello" })); } + + #[tokio::test] + async fn collects_server_stream_messages_with_window_bounds() { + let server_addr = test_support::spawn_unary_echo_server().await; + let adapter = GrpcAdapter::new(); + let target = GrpcTarget { + server_addr, + package: "echo".to_owned(), + service: "EchoService".to_owned(), + method: "ServerEcho".to_owned(), + descriptor_ref: DescriptorId::new("desc_echo"), + descriptor_set_b64: test_support::descriptor_set_b64(), + }; + let request = GrpcWindowRequest { + request: GrpcRequest { + headers: BTreeMap::new(), + body: json!({ "message": "hello" }), + timeout_ms: 1_000, + }, + window_duration_ms: 1_000, + max_items: Some(2), + }; + + let response = adapter.execute_window(&target, &request).await.unwrap(); + + assert_eq!( + response.body, + json!({ + "items": [ + { "message": "hello-one" }, + { "message": "hello-two" } + ], + "done": false + }) + ); + } + + #[tokio::test] + async fn rejects_unary_method_in_window_mode() { + let server_addr = test_support::spawn_unary_echo_server().await; + let adapter = GrpcAdapter::new(); + let target = GrpcTarget { + server_addr, + package: "echo".to_owned(), + service: "EchoService".to_owned(), + method: "UnaryEcho".to_owned(), + descriptor_ref: DescriptorId::new("desc_echo"), + descriptor_set_b64: test_support::descriptor_set_b64(), + }; + let request = GrpcWindowRequest { + request: GrpcRequest { + headers: BTreeMap::new(), + body: json!({ "message": "hello" }), + timeout_ms: 1_000, + }, + window_duration_ms: 1_000, + max_items: Some(10), + }; + + let error = adapter.execute_window(&target, &request).await.unwrap_err(); + + assert!(matches!( + error, + GrpcAdapterError::UnsupportedStreamingMethodKind { .. } + )); + } } diff --git a/crates/crank-adapter-grpc/src/error.rs b/crates/crank-adapter-grpc/src/error.rs index 0cf158a..4ada46d 100644 --- a/crates/crank-adapter-grpc/src/error.rs +++ b/crates/crank-adapter-grpc/src/error.rs @@ -18,8 +18,10 @@ pub enum GrpcAdapterError { MethodNotFound { service: String, method: String }, #[error("grpc method path is invalid for service {service} and method {method}")] InvalidMethodPath { service: String, method: String }, - #[error("grpc method {service}/{method} is not unary")] + #[error("grpc method {service}/{method} is not supported for unary execution")] UnsupportedMethodKind { service: String, method: String }, + #[error("grpc method {service}/{method} does not support server-stream execution")] + UnsupportedStreamingMethodKind { service: String, method: String }, #[error("invalid metadata key {key}")] InvalidMetadataKey { key: String, @@ -34,6 +36,8 @@ pub enum GrpcAdapterError { }, #[error("transport endpoint is invalid")] InvalidEndpoint(#[from] tonic::transport::Error), + #[error("stream collection window expired before grpc stream completed")] + WindowExpired, #[error("grpc status {code}: {message}")] Status { code: Code, message: String }, } diff --git a/crates/crank-adapter-grpc/src/lib.rs b/crates/crank-adapter-grpc/src/lib.rs index fde7905..03812a9 100644 --- a/crates/crank-adapter-grpc/src/lib.rs +++ b/crates/crank-adapter-grpc/src/lib.rs @@ -5,7 +5,7 @@ mod model; pub use client::GrpcAdapter; pub use error::GrpcAdapterError; -pub use model::{GrpcRequest, GrpcResponse}; +pub use model::{GrpcRequest, GrpcResponse, GrpcWindowRequest, GrpcWindowResponse}; #[cfg(any(test, feature = "test-support"))] pub mod test_support; diff --git a/crates/crank-adapter-grpc/src/model.rs b/crates/crank-adapter-grpc/src/model.rs index c04f468..a1a61e1 100644 --- a/crates/crank-adapter-grpc/src/model.rs +++ b/crates/crank-adapter-grpc/src/model.rs @@ -18,3 +18,20 @@ pub struct GrpcResponse { pub headers: BTreeMap, pub body: Value, } + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)] +pub struct GrpcWindowRequest { + #[serde(flatten)] + pub request: GrpcRequest, + pub window_duration_ms: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_items: Option, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct GrpcWindowResponse { + pub status_code: u16, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub headers: BTreeMap, + pub body: Value, +} diff --git a/crates/crank-adapter-grpc/src/test_support.rs b/crates/crank-adapter-grpc/src/test_support.rs index 4ad7dfd..c0cae13 100644 --- a/crates/crank-adapter-grpc/src/test_support.rs +++ b/crates/crank-adapter-grpc/src/test_support.rs @@ -1,4 +1,5 @@ use base64::{Engine as _, engine::general_purpose::STANDARD}; +use futures_util::stream; use tokio::net::TcpListener; use tonic::{Request, Response, Status, transport::Server}; @@ -20,6 +21,30 @@ impl echo::echo_service_server::EchoService for EchoServiceImpl { message: request.into_inner().message, })) } + + type ServerEchoStream = std::pin::Pin< + Box> + Send>, + >; + + async fn server_echo( + &self, + request: Request, + ) -> Result, Status> { + let message = request.into_inner().message; + let events = vec![ + Ok(echo::EchoResponse { + message: format!("{message}-one"), + }), + Ok(echo::EchoResponse { + message: format!("{message}-two"), + }), + Ok(echo::EchoResponse { + message: format!("{message}-three"), + }), + ]; + + Ok(Response::new(Box::pin(stream::iter(events)))) + } } pub async fn spawn_unary_echo_server() -> String { diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index 1611c37..c4b2748 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -1,7 +1,7 @@ use std::collections::BTreeMap; use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest}; -use crank_adapter_grpc::{GrpcAdapter, GrpcRequest}; +use crank_adapter_grpc::{GrpcAdapter, GrpcRequest, GrpcWindowRequest}; use crank_adapter_rest::{RestAdapter, RestRequest, RestWindowRequest}; use crank_core::{ExecutionMode, Target, TransportBehavior}; use serde_json::{Map, Value, json}; @@ -63,7 +63,7 @@ impl RuntimeExecutor { let adapter_response = if matches!( streaming.transport_behavior, TransportBehavior::ServerStream - ) && matches!(operation.target, Target::Rest(_)) + ) && matches!(operation.target, Target::Rest(_) | Target::Grpc(_)) { self.execute_window_adapter(operation, prepared_request) .await? @@ -207,6 +207,39 @@ impl RuntimeExecutor { data: Value::Null, }) } + Target::Grpc(target) => { + let Some(streaming) = operation.execution_config.streaming.as_ref() else { + return Err(RuntimeError::MissingStreamingConfig { + operation_id: operation.operation_id.as_str().to_owned(), + }); + }; + let request = GrpcWindowRequest { + request: GrpcRequest { + headers: merge_headers( + &BTreeMap::new(), + &operation.execution_config.headers, + &prepared_request.headers, + ), + body: prepared_request + .grpc + .clone() + .unwrap_or(Value::Object(Map::new())), + timeout_ms: streaming + .upstream_timeout_ms + .unwrap_or(operation.execution_config.timeout_ms), + }, + window_duration_ms: streaming.window_duration_ms.unwrap_or_default(), + max_items: streaming.max_items.map(|value| value.saturating_add(1)), + }; + let response = self.grpc_adapter.execute_window(target, &request).await?; + + Ok(AdapterResponse { + status_code: response.status_code, + headers: response.headers, + body: response.body, + data: Value::Null, + }) + } _ => self.execute_adapter(operation, prepared_request).await, } } @@ -377,6 +410,26 @@ mod tests { assert_eq!(output, json!({ "message": "hello" })); } + #[tokio::test] + async fn executes_grpc_window_mode_with_server_stream() { + let server_addr = grpc_test_support::spawn_unary_echo_server().await; + let executor = RuntimeExecutor::new(); + let operation = + test_grpc_window_operation(&server_addr, AggregationMode::RawItems, Some(2)); + + let result = executor + .execute_window(&operation, &json!({ "message": "hello" })) + .await + .unwrap(); + + assert_eq!(result.summary, Value::Null); + assert_eq!(result.items.len(), 2); + assert_eq!(result.items[0], json!({ "message": "hello-one" })); + assert!(result.truncated); + assert!(result.has_more); + assert!(!result.window_complete); + } + #[tokio::test] async fn rejects_invalid_input_shape() { let base_url = spawn_runtime_server().await; @@ -898,6 +951,88 @@ mod tests { }) } + fn test_grpc_window_operation( + server_addr: &str, + aggregation_mode: AggregationMode, + max_items: Option, + ) -> RuntimeOperation { + RuntimeOperation::from(Operation { + id: OperationId::new("op_grpc_window_runtime"), + name: "echo_stream_grpc".to_owned(), + display_name: "Server Stream Echo gRPC".to_owned(), + category: "support".to_owned(), + protocol: Protocol::Grpc, + status: OperationStatus::Published, + version: 1, + target: Target::Grpc(GrpcTarget { + server_addr: server_addr.to_owned(), + package: "echo".to_owned(), + service: "EchoService".to_owned(), + method: "ServerEcho".to_owned(), + descriptor_ref: DescriptorId::new("desc_echo"), + descriptor_set_b64: grpc_test_support::descriptor_set_b64(), + }), + input_schema: object_schema("message", SchemaKind::String), + output_schema: object_schema("ignored", SchemaKind::String), + input_mapping: MappingSet { + rules: vec![MappingRule { + source: "$.mcp.message".to_owned(), + target: "$.request.grpc.message".to_owned(), + required: true, + default_value: None, + transform: None, + condition: None, + notes: None, + }], + }, + output_mapping: MappingSet { rules: Vec::new() }, + execution_config: ExecutionConfig { + timeout_ms: 1_000, + retry_policy: None, + auth_profile_ref: None, + headers: BTreeMap::new(), + protocol_options: None, + streaming: Some(StreamingConfig { + mode: ExecutionMode::Window, + transport_behavior: TransportBehavior::ServerStream, + window_duration_ms: Some(3_000), + poll_interval_ms: None, + upstream_timeout_ms: Some(1_000), + idle_timeout_ms: None, + max_session_lifetime_ms: None, + max_items, + max_bytes: None, + aggregation_mode, + summary_path: None, + items_path: Some("$.items".to_owned()), + cursor_path: None, + status_path: None, + done_path: Some("$.done".to_owned()), + redacted_paths: Vec::new(), + truncate_item_fields: false, + max_field_length: None, + drop_duplicates: false, + sampling_rate: None, + tool_family: ToolFamilyConfig::default(), + }), + }, + tool_description: ToolDescription { + title: "Server Stream Echo gRPC".to_owned(), + description: "Collects bounded gRPC stream messages.".to_owned(), + tags: vec!["grpc".to_owned(), "stream".to_owned()], + examples: vec![ToolExample { + input: json!({ "message": "hello" }), + }], + }, + samples: None, + generated_draft: None, + config_export: None, + created_at: "2026-04-06T12:00:00Z".to_owned(), + updated_at: "2026-04-06T12:00:00Z".to_owned(), + published_at: Some("2026-04-06T12:00:00Z".to_owned()), + }) + } + fn test_window_snapshot_operation( base_url: &str, aggregation_mode: AggregationMode,