sui_rpc_api/grpc/
mod.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::convert::Infallible;
5use std::time::Duration;
6use sui_http::middleware::grpc_timeout::GrpcTimeout;
7use tonic::server::NamedService;
8use tower::Service;
9use tower::layer::layer_fn;
10
11pub mod deadline;
12pub mod v2;
13pub mod v2alpha;
14
15pub type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
16
17#[derive(Default)]
18pub struct Services {
19    router: axum::Router,
20    timeout: Option<Duration>,
21}
22
23impl Services {
24    pub fn new() -> Self {
25        Self::default()
26    }
27
28    /// Server-side deadline applied to every gRPC request mounted here.
29    /// Requests carrying a `grpc-timeout` header are bounded by the smaller
30    /// of the two values. The deadline covers a unary request's execution
31    /// and a streaming request's time to first response; it does not bound
32    /// the lifetime of an established stream.
33    pub fn timeout(mut self, timeout: Option<Duration>) -> Self {
34        self.timeout = timeout;
35        self
36    }
37
38    /// Add a new service.
39    pub fn add_service<S>(mut self, svc: S) -> Self
40    where
41        S: Service<
42                axum::extract::Request,
43                Response: axum::response::IntoResponse,
44                Error = Infallible,
45            > + NamedService
46            + Clone
47            + Send
48            + Sync
49            + 'static,
50        S::Future: Send + 'static,
51        S::Error: Into<BoxError> + Send,
52    {
53        self.router = self
54            .router
55            .route_service(&format!("/{}/{{*rest}}", S::NAME), svc);
56        self
57    }
58
59    pub fn merge_router(mut self, router: axum::Router) -> Self {
60        self.router = self.router.merge(router);
61        self
62    }
63
64    pub fn into_router(
65        self,
66        request_log: mysten_network::request_log::GrpcRequestLogLayer,
67    ) -> axum::Router {
68        let timeout = self.timeout;
69        self.router
70            // The capture layer sits under `GrpcWebLayer` (the last layer added is outermost) so
71            // it always sees standard gRPC frames, including for grpc-web(-text) requests.
72            .layer(request_log)
73            // The timeout sits inside the grpc-web layer so that its
74            // trailers-only DeadlineExceeded response is translated for
75            // grpc-web clients too.
76            .layer(layer_fn(move |service| GrpcTimeout::new(service, timeout)))
77            .layer(tonic_web::GrpcWebLayer::new())
78    }
79}
80
81#[cfg(test)]
82mod tests {
83    use std::future::Future;
84    use std::pin::Pin;
85    use std::sync::Arc;
86    use std::sync::Mutex;
87    use std::task::Context;
88    use std::task::Poll;
89
90    use base64::Engine as _;
91    use mysten_network::request_log::GrpcRequestLogLayer;
92    use prost::Message;
93    use tower::ServiceExt;
94    use tracing_subscriber::layer::SubscriberExt;
95
96    use super::*;
97
98    /// A gRPC service whose handler never completes, standing in for a
99    /// request wedged on a lock, a stalled backend, or an h2 send window
100    /// that never reopens.
101    #[derive(Clone)]
102    struct HangingService;
103
104    impl Service<axum::extract::Request> for HangingService {
105        type Response = axum::response::Response;
106        type Error = Infallible;
107        type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
108
109        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
110            Poll::Ready(Ok(()))
111        }
112
113        fn call(&mut self, _request: axum::extract::Request) -> Self::Future {
114            Box::pin(std::future::pending())
115        }
116    }
117
118    impl NamedService for HangingService {
119        const NAME: &'static str = "test.Hanging";
120    }
121
122    fn request(grpc_timeout: Option<&str>) -> axum::extract::Request {
123        let mut builder = http::Request::builder()
124            .method(http::Method::POST)
125            .uri("/test.Hanging/Method")
126            // The grpc-web layer only passes through native gRPC requests
127            // arriving over HTTP/2.
128            .version(http::Version::HTTP_2)
129            .header(http::header::CONTENT_TYPE, "application/grpc");
130        if let Some(timeout) = grpc_timeout {
131            builder = builder.header("grpc-timeout", timeout);
132        }
133        builder.body(axum::body::Body::empty()).unwrap()
134    }
135
136    fn grpc_status(response: &http::Response<axum::body::Body>) -> Option<&str> {
137        response
138            .headers()
139            .get("grpc-status")
140            .and_then(|value| value.to_str().ok())
141    }
142
143    /// A request-log layer with an empty descriptor pool: `capture_state` never resolves a
144    /// service/method against it, so it's a pure pass-through — for tests that exercise
145    /// unrelated `Services` behavior and don't care about capture.
146    fn empty_request_log() -> GrpcRequestLogLayer {
147        GrpcRequestLogLayer::from_encoded_file_descriptor_sets([]).unwrap()
148    }
149
150    /// The server-side default deadline must bound a request whose handler
151    /// never completes, surfacing gRPC status 4 (DeadlineExceeded) instead
152    /// of hanging the client forever.
153    #[tokio::test(start_paused = true)]
154    async fn server_timeout_bounds_a_hung_handler() {
155        let router = Services::new()
156            .timeout(Some(Duration::from_millis(50)))
157            .add_service(HangingService)
158            .into_router(empty_request_log());
159
160        let response = router.oneshot(request(None)).await.unwrap();
161        assert_eq!(response.status(), http::StatusCode::OK);
162        assert_eq!(grpc_status(&response), Some("4"));
163    }
164
165    /// A client-supplied `grpc-timeout` header must be honored even when no
166    /// server default is configured.
167    #[tokio::test(start_paused = true)]
168    async fn client_grpc_timeout_header_is_honored() {
169        let router = Services::new()
170            .timeout(None)
171            .add_service(HangingService)
172            .into_router(empty_request_log());
173
174        let response = router.oneshot(request(Some("50m"))).await.unwrap();
175        assert_eq!(response.status(), http::StatusCode::OK);
176        assert_eq!(grpc_status(&response), Some("4"));
177    }
178
179    /// With no server default and no client header there is no deadline:
180    /// the request must still be pending well past any implicit bound. This
181    /// pins the disabled behavior (config value `0`).
182    #[tokio::test(start_paused = true)]
183    async fn no_timeout_means_no_deadline() {
184        let router = Services::new()
185            .timeout(None)
186            .add_service(HangingService)
187            .into_router(empty_request_log());
188
189        let response = tokio::time::timeout(
190            Duration::from_secs(24 * 60 * 60),
191            router.oneshot(request(None)),
192        )
193        .await;
194        assert!(response.is_err(), "request completed without a deadline");
195    }
196
197    /// Records the `payload` field of every `grpc_request` event.
198    #[derive(Clone, Default)]
199    struct CaptureLayer {
200        payloads: Arc<Mutex<Vec<String>>>,
201    }
202
203    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CaptureLayer {
204        fn on_event(
205            &self,
206            event: &tracing::Event<'_>,
207            _ctx: tracing_subscriber::layer::Context<'_, S>,
208        ) {
209            struct Visitor(Option<String>);
210            impl tracing::field::Visit for Visitor {
211                fn record_debug(
212                    &mut self,
213                    field: &tracing::field::Field,
214                    value: &dyn std::fmt::Debug,
215                ) {
216                    if field.name() == "payload" {
217                        self.0 = Some(format!("{value:?}"));
218                    }
219                }
220            }
221
222            let mut visitor = Visitor(None);
223            event.record(&mut visitor);
224            if let Some(payload) = visitor.0 {
225                self.payloads.lock().unwrap().push(payload);
226            }
227        }
228    }
229
230    /// The request-log layer must sit *under* `GrpcWebLayer` so it sees standard gRPC frames for
231    /// grpc-web requests too. If the ordering regresses, the capture stream silently loses all
232    /// browser-client traffic.
233    #[tokio::test]
234    async fn request_log_captures_grpc_web_requests() {
235        let capture_layer = CaptureLayer::default();
236        let subscriber = tracing_subscriber::registry()
237            .with(tracing_subscriber::EnvFilter::new("grpc_request=trace"))
238            .with(capture_layer.clone());
239        let _guard = tracing::subscriber::set_default(subscriber);
240
241        let (_health_reporter, health_service) = tonic_health::server::health_reporter();
242        let router = Services::new().add_service(health_service).into_router(
243            GrpcRequestLogLayer::from_encoded_file_descriptor_sets([
244                tonic_health::pb::FILE_DESCRIPTOR_SET,
245            ])
246            .unwrap(),
247        );
248
249        let message = tonic_health::pb::HealthCheckRequest {
250            service: "x".to_owned(),
251        }
252        .encode_to_vec();
253        let mut body = vec![0u8];
254        body.extend_from_slice(&(message.len() as u32).to_be_bytes());
255        body.extend_from_slice(&message);
256
257        let request = axum::http::Request::builder()
258            .method(axum::http::Method::POST)
259            .uri("/grpc.health.v1.Health/Check")
260            .header(
261                axum::http::header::CONTENT_TYPE,
262                "application/grpc-web+proto",
263            )
264            .body(axum::body::Body::from(body))
265            .unwrap();
266
267        let response = router.oneshot(request).await.unwrap();
268        assert_eq!(response.status(), axum::http::StatusCode::OK);
269
270        let payloads = capture_layer.payloads.lock().unwrap();
271        assert_eq!(
272            *payloads,
273            vec![base64::engine::general_purpose::STANDARD.encode(&message)]
274        );
275    }
276}
sui_rpc_api/grpc/mod.rs

sui_rpc_api/grpc/
mod.rs
