grpc_webnext/
lib.rs

1//! grpc-webnext: serve the grpc-webnext wire protocol (unary over Fetch, streaming over
2//! WebSocket, plus `+json` / REST transcoding) over any gRPC service.
3//!
4//! All the inbound protocol translation is shared; the only thing that varies is where the
5//! translated gRPC call lands — a local in-process [`tonic::service::Routes`] you own
6//! ([`serve_in_process`], the "native server" / wrap mode) or a remote upstream over an
7//! HTTP/2 channel ([`serve_proxy`], the standalone binary proxy). Both are the same
8//! [`Backend`]; the two entry points build one and run the identical handlers, so a client
9//! can't tell a wrapped response from a proxied one.
10
11use std::future::Future;
12use std::net::SocketAddr;
13use std::sync::Arc;
14use std::time::Duration;
15
16use bytes::Bytes;
17use http::HeaderMap;
18use http_body_util::combinators::UnsyncBoxBody;
19use hyper::body::Incoming;
20use hyper::Request;
21use hyper_util::rt::{TokioExecutor, TokioIo};
22use tokio::net::TcpListener;
23use tonic::service::Routes;
24use tonic::transport::Channel;
25
26// Inbound protocol translation.
27pub mod backend;
28mod drain;
29mod fetch;
30mod h2ts;
31mod reflect;
32pub mod schema;
33mod ws;
34
35// Wire codec + gRPC-semantics core: generated types, frame codec, Fetch-response framing,
36// metadata, HTTP-rule transcoding.
37pub mod pb {
38    include!(concat!(env!("OUT_DIR"), "/grpc.webnext.v1.rs"));
39}
40pub mod codec;
41pub mod frame;
42mod framing;
43pub mod grpc_framing;
44pub mod httprule;
45pub mod json_frame;
46pub mod metadata;
47pub mod transcode;
48
49pub(crate) use drain::Drain;
50
51pub use backend::Backend;
52pub use schema::{Schema, SchemaSource};
53
54pub use codec::BytesCodec;
55pub use frame::{decode_frame, encode_frame, FrameError};
56pub use framing::{
57    decode_response_body, encode_request_body, encode_response_body, encode_trailer_block,
58    FetchError, EMPTY_MESSAGE_BLOCK, LEN_PREFIX,
59};
60pub use grpc_framing::{deframe_all, frame as grpc_frame, Deframer};
61pub use httprule::{HttpCall, HttpRouter, WsBinding};
62pub use transcode::{TranscodeError, Transcoder};
63
64// --- Wire constants ---------------------------------------------------------
65
66pub const CT_PROTO: &str = "application/grpc-webnext+proto";
67pub const CT_JSON: &str = "application/grpc-webnext+json";
68pub(crate) const CT_GRPC: &str = "application/grpc";
69
70/// Base subprotocol; a client offers it plus a codec/credential entry.
71pub const WS_SUBPROTOCOL: &str = "grpc-webnext";
72pub const WS_SUBPROTOCOL_JSON: &str = "grpc-webnext+json";
73pub const WS_SUBPROTOCOL_PROTO: &str = "grpc-webnext+proto";
74
75/// The proxy owns the client-facing deadline: it drops the call at the deadline
76/// (surfacing DEADLINE_EXCEEDED) and forwards `grpc-timeout` downstream with this grace so
77/// the callee's own enforcement is a later backstop rather than racing the local timer.
78pub const DEADLINE_GRACE: Duration = Duration::from_millis(500);
79
80pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
81pub type ResBody = UnsyncBoxBody<Bytes, BoxError>;
82
83// Authorization is deliberately NOT a grpc-webnext hook — neither per-RPC nor
84// per-connection. The request carries its metadata to the router on every transport, so
85// auth belongs in a tonic interceptor (in-process) or the upstream server / mesh (proxy):
86// one uniform, per-RPC check that also covers the native/h2ts path. Connection-level app
87// auth is non-canonical (the ecosystem gates connections only on network identity, e.g.
88// mTLS) and can't be uniform (Fetch has no connection). See `tests/inproc_auth.rs`.
89
90// --- Public configs ---------------------------------------------------------
91
92/// Configuration for [`serve_in_process`] — wrap an in-process tonic service.
93#[derive(Clone)]
94pub struct ServerConfig {
95    pub max_message_bytes: usize,
96    /// Descriptor-based JSON<->proto transcoder. When set, `+json`/REST requests are
97    /// transcoded to the router's binary protobuf and back. `None` ⇒ `+json` is
98    /// `UNIMPLEMENTED`.
99    pub transcoder: Option<Arc<Transcoder>>,
100    /// Allow plain `application/json` / blank content-type to reach *main* gRPC paths
101    /// (and blank WS subprotocols to infer their codec). Off by default.
102    pub allow_implicit_codec: bool,
103    /// WebSocket keepalive ping interval (`None` disables).
104    pub ws_keepalive: Option<Duration>,
105    /// Dead-peer timeout after a keepalive ping (gRPC's `keepalive_timeout`, default 20s).
106    pub ws_keepalive_timeout: Duration,
107}
108
109impl Default for ServerConfig {
110    fn default() -> Self {
111        Self {
112            max_message_bytes: 4 * 1024 * 1024,
113            transcoder: None,
114            allow_implicit_codec: false,
115            ws_keepalive: None,
116            ws_keepalive_timeout: Duration::from_secs(20),
117        }
118    }
119}
120
121/// Configuration for [`serve_proxy`] — front a remote upstream gRPC server.
122#[derive(Clone)]
123pub struct ProxyConfig {
124    /// Upstream gRPC endpoint (e.g. `http://127.0.0.1:50051`).
125    pub upstream: http::Uri,
126    pub max_message_bytes: usize,
127    pub ws_keepalive: Option<Duration>,
128    pub ws_keepalive_timeout: Duration,
129    /// Descriptor source for `+json` termination (`None` ⇒ binary-only).
130    pub schema: SchemaSource,
131    /// Reflection snapshot refresh interval (default 4h).
132    pub reflection_ttl: Duration,
133    /// Optional management endpoint: `POST` to this exact path forces a reflection reload.
134    pub admin_reload_path: Option<String>,
135    /// Accept plain `application/json` / blank on *main* gRPC paths (off by default).
136    pub allow_implicit_codec: bool,
137}
138
139impl Default for ProxyConfig {
140    fn default() -> Self {
141        Self {
142            upstream: http::Uri::default(),
143            max_message_bytes: 4 * 1024 * 1024,
144            ws_keepalive: None,
145            ws_keepalive_timeout: Duration::from_secs(20),
146            schema: SchemaSource::None,
147            reflection_ttl: Duration::from_secs(4 * 60 * 60),
148            admin_reload_path: None,
149            allow_implicit_codec: false,
150        }
151    }
152}
153
154// --- Internal runtime -------------------------------------------------------
155
156/// The per-connection knobs the shared handlers read, independent of which surface built
157/// them.
158pub(crate) struct RunConfig {
159    pub max_message_bytes: usize,
160    pub ws_keepalive: Option<Duration>,
161    pub ws_keepalive_timeout: Duration,
162    pub allow_implicit_codec: bool,
163    pub admin_reload_path: Option<String>,
164    /// Upstream `host:port` for the h2ts byte-pump path (proxy only; `None` in-process).
165    pub upstream_authority: Option<String>,
166}
167
168/// Everything a connection needs: where to dispatch, how to transcode, and the policy.
169/// Cheap to clone.
170///
171/// Holding one also makes a task a *live connection* for shutdown purposes — the
172/// [`Drain`] inside carries the liveness token a drain waits on — so a task that
173/// outlives its connection must not keep a clone.
174#[derive(Clone)]
175pub(crate) struct Runtime {
176    pub backend: Backend,
177    pub schema: Schema,
178    pub cfg: Arc<RunConfig>,
179    pub drain: Drain,
180}
181
182// --- Entry points -----------------------------------------------------------
183
184/// Serve grpc-webnext + native gRPC for an in-process `routes` on `listener`, until the
185/// listener errors.
186pub async fn serve_in_process(
187    listener: TcpListener,
188    routes: Routes,
189    config: ServerConfig,
190) -> std::io::Result<()> {
191    serve_in_process_with_shutdown(listener, routes, config, std::future::pending()).await
192}
193
194/// [`serve_in_process`], but draining once `shutdown` resolves.
195///
196/// Draining stops accepting connections, tells every open one to finish (an HTTP/2
197/// `GOAWAY` — including down an in-process h2ts tunnel — and a close on any WebSocket
198/// that has not opened its stream yet), lets in-flight RPCs complete, and resolves when
199/// the last connection is gone.
200///
201/// There is deliberately no drain-timeout knob: the returned future *is* the drain, so
202/// bounding it is [`tokio::time::timeout`], and dropping it force-closes whatever is
203/// still open. A long-lived stream will hold the drain open until then, which is what
204/// "graceful" means.
205///
206/// ```no_run
207/// # async fn f(listener: tokio::net::TcpListener, routes: tonic::service::Routes) {
208/// use std::time::Duration;
209/// let server = grpc_webnext::serve_in_process_with_shutdown(
210///     listener,
211///     routes,
212///     Default::default(),
213///     async { tokio::signal::ctrl_c().await.ok(); },
214/// );
215/// // Drain, but never take longer than 30s to exit.
216/// let _ = tokio::time::timeout(Duration::from_secs(30), server).await;
217/// # }
218/// ```
219pub async fn serve_in_process_with_shutdown(
220    listener: TcpListener,
221    routes: Routes,
222    config: ServerConfig,
223    shutdown: impl Future<Output = ()>,
224) -> std::io::Result<()> {
225    let schema = Schema::from_transcoder(config.transcoder.clone());
226    let cfg = Arc::new(RunConfig {
227        max_message_bytes: config.max_message_bytes,
228        ws_keepalive: config.ws_keepalive,
229        ws_keepalive_timeout: config.ws_keepalive_timeout,
230        allow_implicit_codec: config.allow_implicit_codec,
231        admin_reload_path: None,
232        upstream_authority: None,
233    });
234    let (controller, drain) = drain::channel();
235    let rt = Runtime { backend: Backend::InProcess(routes), schema, cfg, drain };
236    run(listener, rt, controller, shutdown).await
237}
238
239/// Serve grpc-webnext + native gRPC passthrough in front of an upstream gRPC server.
240pub async fn serve_proxy(listener: TcpListener, config: ProxyConfig) -> std::io::Result<()> {
241    serve_proxy_with_shutdown(listener, config, std::future::pending()).await
242}
243
244/// [`serve_proxy`], but draining once `shutdown` resolves — see
245/// [`serve_in_process_with_shutdown`] for the semantics.
246///
247/// One surface drains differently here: the proxy's h2ts tunnels are a **byte-transparent**
248/// pump to the upstream, so there is no `GOAWAY` to inject without parsing the very traffic
249/// the proxy exists not to parse. Those tunnels run until the caller's deadline; every other
250/// surface drains as it does in-process.
251pub async fn serve_proxy_with_shutdown(
252    listener: TcpListener,
253    config: ProxyConfig,
254    shutdown: impl Future<Output = ()>,
255) -> std::io::Result<()> {
256    // Lazy connect: the upstream need not be up when we start.
257    let channel = Channel::builder(config.upstream.clone()).connect_lazy();
258    // Raw host:port for the h2ts byte-pump path (which bypasses the gRPC `Channel`).
259    let upstream_authority = config.upstream.authority().map(|a| match a.port_u16() {
260        Some(_) => a.to_string(),
261        None => {
262            let port = if config.upstream.scheme_str() == Some("https") { 443 } else { 80 };
263            format!("{}:{}", a.host(), port)
264        }
265    });
266    // A bad bundled descriptor set is a config error — surface it at startup.
267    let schema = Schema::build(config.schema.clone(), channel.clone(), config.reflection_ttl)
268        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e.to_string()))?;
269    // Eager reflection load + TTL refresh (no-op for None/Bundled).
270    schema.start();
271    let cfg = Arc::new(RunConfig {
272        max_message_bytes: config.max_message_bytes,
273        ws_keepalive: config.ws_keepalive,
274        ws_keepalive_timeout: config.ws_keepalive_timeout,
275        allow_implicit_codec: config.allow_implicit_codec,
276        admin_reload_path: config.admin_reload_path,
277        upstream_authority,
278    });
279    let (controller, drain) = drain::channel();
280    let rt = Runtime { backend: Backend::Upstream(channel), schema, cfg, drain };
281    run(listener, rt, controller, shutdown).await
282}
283
284/// The connection accept loop, shared by both surfaces.
285async fn run(
286    listener: TcpListener,
287    rt: Runtime,
288    controller: drain::DrainController,
289    shutdown: impl Future<Output = ()>,
290) -> std::io::Result<()> {
291    let mut shutdown = std::pin::pin!(shutdown);
292    let result = loop {
293        let accepted = tokio::select! {
294            accepted = listener.accept() => accepted,
295            _ = shutdown.as_mut() => break Ok(()),
296        };
297        let (stream, _peer) = match accepted {
298            Ok(pair) => pair,
299            Err(e) => break Err(e),
300        };
301        let io = TokioIo::new(stream);
302        let rt = rt.clone();
303        tokio::spawn(async move {
304            let drain = rt.drain.clone();
305            let service = hyper::service::service_fn(move |req: Request<Incoming>| {
306                let rt = rt.clone();
307                async move { fetch::handle(&rt, req).await }
308            });
309            // The connection borrows the builder, so the builder has to outlive it.
310            let builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
311            let conn = builder.serve_connection_with_upgrades(io, service);
312            // Once draining starts this asks the connection to finish: `GOAWAY` on
313            // HTTP/2, no-more-requests on HTTP/1. In-flight requests still complete.
314            if let Err(e) = drain::serve_graceful!(&drain, conn) {
315                tracing::debug!("connection error: {e}");
316            }
317        });
318    };
319
320    // Dropping the listener stops accepting; dropping `rt` releases this task's own
321    // liveness token, without which the drain would wait on itself forever.
322    drop(listener);
323    drop(rt);
324    controller.finish().await;
325    result
326}
327
328/// Bind an ephemeral local address, serve an in-process `routes`, and return the bound
329/// address + task handle. For tests and simple mains.
330pub async fn bind_and_serve_in_process(
331    routes: Routes,
332    config: ServerConfig,
333) -> std::io::Result<(SocketAddr, tokio::task::JoinHandle<std::io::Result<()>>)> {
334    let listener = TcpListener::bind("127.0.0.1:0").await?;
335    let addr = listener.local_addr()?;
336    let handle = tokio::spawn(serve_in_process(listener, routes, config));
337    Ok((addr, handle))
338}
339
340/// Bind an ephemeral local address, serve a proxy, and return the bound address + handle.
341pub async fn bind_and_serve_proxy(
342    config: ProxyConfig,
343) -> std::io::Result<(SocketAddr, tokio::task::JoinHandle<std::io::Result<()>>)> {
344    let listener = TcpListener::bind("127.0.0.1:0").await?;
345    let addr = listener.local_addr()?;
346    let handle = tokio::spawn(serve_proxy(listener, config));
347    Ok((addr, handle))
348}
349
350// --- WebSocket handshake helpers --------------------------------------------
351
352/// Parse the `Sec-WebSocket-Protocol` request header into its comma-separated tokens.
353/// Used to negotiate the codec / detect the `h2ts` subprotocol at handshake time.
354pub fn ws_subprotocols(headers: &HeaderMap) -> Vec<String> {
355    headers
356        .get(http::header::SEC_WEBSOCKET_PROTOCOL)
357        .and_then(|v| v.to_str().ok())
358        .map(|s| s.split(',').map(|t| t.trim().to_string()).filter(|t| !t.is_empty()).collect())
359        .unwrap_or_default()
360}
grpc_webnext/lib.rs

grpc_webnext/
lib.rs
