feat: add grpc server streaming adapter

2026-04-06 11:04:10 +03:00
parent bf56494336
commit bd2c6d4f48
10 changed files with 358 additions and 31 deletions
@@ -363,6 +363,7 @@ dependencies = [
 "base64",
 "crank-core",
 "crank-proto",
+ "futures-util",
 "prost",
 "prost-reflect",
 "prost-types",
@@ -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

@@ -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
@@ -4,6 +4,7 @@ package echo;

 service EchoService {
  rpc UnaryEcho(EchoRequest) returns (EchoResponse);
+  rpc ServerEcho(EchoRequest) returns (stream EchoResponse);
 }

 message EchoRequest {
@@ -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<GrpcWindowResponse, GrpcAdapterError> {
+        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<Grpc<tonic::transport::Channel>, 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<tonic::codegen::http::uri::PathAndQuery, GrpcAdapterError> {
+    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<MethodDescriptor, GrpcAdapterError> {
@@ -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 { .. }
+        ));
+    }
 }
@@ -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 },
 }
@@ -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;
@@ -18,3 +18,20 @@ pub struct GrpcResponse {
    pub headers: BTreeMap<String, String>,
    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<u32>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct GrpcWindowResponse {
+    pub status_code: u16,
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub headers: BTreeMap<String, String>,
+    pub body: Value,
+}
@@ -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<dyn futures_util::Stream<Item = Result<echo::EchoResponse, Status>> + Send>,
+    >;
+
+    async fn server_echo(
+        &self,
+        request: Request<echo::EchoRequest>,
+    ) -> Result<Response<Self::ServerEchoStream>, 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 {
@@ -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<u32>,
+    ) -> 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,