netget/server/connect_rpc/
mod.rs

1//! Native binary protobuf Connect over owned HTTP/1.1 connections.
2pub mod actions;
3pub mod wire;
4use crate::server::grpc::{
5    self,
6    streaming::{ConnectionTasks, OwnedExecutor, Registry},
7    HttpCore,
8};
9use crate::{
10    protocol::SpawnContext,
11    state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo},
12};
13use anyhow::{ensure, Context, Result};
14use http_body_util::BodyExt;
15use hyper::{body::Incoming, Method, Request, Response, StatusCode};
16use std::{convert::Infallible, sync::Arc, time::Duration};
17use tonic::{body::BoxBody, Status};
18pub const FIRST_BYTE_TIMEOUT: Duration = Duration::from_secs(30);
19pub const IDLE_TIMEOUT: Duration = Duration::from_secs(120);
20pub const MAX_CONNECTIONS: usize = 256;
21pub const DEFAULT_RPC_TIMEOUT_SECS: u64 = 300;
22const REFUSAL:&[u8]=b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nRetry-After: 5\r\nConnection: close\r\n\r\n";
23fn refusal(code: StatusCode, status: Status) -> Response<BoxBody> {
24    let mut response = wire::failure(status, wire::Shape::Unary, true);
25    *response.status_mut() = code;
26    response
27}
28async fn request(
29    request: Request<Incoming>,
30    core: HttpCore,
31    connection: crate::server::connection::ConnectionId,
32    activity: Arc<crate::server::accept_bounded::ConnectionActivity>,
33    registry: Registry,
34    executor: OwnedExecutor,
35) -> Response<BoxBody> {
36    if !wire::bounded_headers(request.headers()) {
37        return refusal(
38            StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE,
39            Status::resource_exhausted("HTTP headers exceed bound"),
40        );
41    }
42    if request.headers().contains_key("origin") {
43        return refusal(
44            StatusCode::FORBIDDEN,
45            Status::permission_denied("browser origin outside native Connect scope"),
46        );
47    }
48    if request.method() != Method::POST {
49        return refusal(
50            StatusCode::METHOD_NOT_ALLOWED,
51            Status::unimplemented("Connect POST only; GET excluded"),
52        );
53    }
54    if request.uri().path().len() > 2048 || request.uri().query().is_some() {
55        return refusal(
56            StatusCode::BAD_REQUEST,
57            Status::invalid_argument("RPC path exceeds bound or has query"),
58        );
59    }
60    let shape = match wire::shape(request.headers()) {
61        Ok(shape) => shape,
62        Err(status) => return refusal(StatusCode::UNSUPPORTED_MEDIA_TYPE, status),
63    };
64    let request_gzip = match wire::encoding(
65        request.headers(),
66        if shape == wire::Shape::Unary {
67            "content-encoding"
68        } else {
69            "connect-content-encoding"
70        },
71    ) {
72        Ok(value) => value,
73        Err(status) => return wire::failure(status, shape, true),
74    };
75    let accept_gzip = match wire::accepts_gzip(
76        request.headers(),
77        if shape == wire::Shape::Unary {
78            "accept-encoding"
79        } else {
80            "connect-accept-encoding"
81        },
82        request_gzip,
83    ) {
84        Ok(value) => value,
85        Err(status) => return wire::failure(status, shape, true),
86    };
87    let timeout = match wire::timeout_ms(request.headers()) {
88        Ok(value) => value,
89        Err(status) => return wire::failure(status, shape, true),
90    };
91    if wire::unique(request.headers(), "connect-protocol-version")
92        .ok()
93        .flatten()
94        != Some("1")
95    {
96        return refusal(
97            StatusCode::BAD_REQUEST,
98            Status::invalid_argument("unique connect-protocol-version: 1 required"),
99        );
100    }
101    if request
102        .headers()
103        .keys()
104        .any(|key| key.as_str().starts_with("grpc-") || key.as_str().ends_with("-bin"))
105        || (shape == wire::Shape::Stream && request.headers().contains_key("content-encoding"))
106        || (shape == wire::Shape::Unary
107            && (request.headers().contains_key("connect-content-encoding")
108                || request.headers().contains_key("connect-accept-encoding")))
109    {
110        return wire::failure(
111            Status::invalid_argument("reserved or unsupported transport headers"),
112            shape,
113            true,
114        );
115    }
116    let metadata = match wire::RpcMetadata::new(request.headers()) {
117        Ok(value) => value,
118        Err(status) => return wire::failure(status, shape, true),
119    };
120    let busy = activity.busy();
121    let (mut parts, body) = request.into_parts();
122    for name in [
123        "content-type",
124        "content-encoding",
125        "accept-encoding",
126        "connect-content-encoding",
127        "connect-accept-encoding",
128        "connect-protocol-version",
129        "connect-timeout-ms",
130    ] {
131        parts.headers.remove(name);
132    }
133    parts.headers.insert(
134        "content-type",
135        http::HeaderValue::from_static("application/grpc"),
136    );
137    if request_gzip && shape == wire::Shape::Stream {
138        parts
139            .headers
140            .insert("grpc-encoding", http::HeaderValue::from_static("gzip"));
141    }
142    if accept_gzip {
143        parts.headers.insert(
144            "grpc-accept-encoding",
145            http::HeaderValue::from_static("gzip"),
146        );
147    }
148    if let Some(timeout) = timeout {
149        parts.headers.insert(
150            "grpc-timeout",
151            http::HeaderValue::from_str(&format!("{}m", timeout.min(3_600_000))).unwrap(),
152        );
153    }
154    parts.extensions.insert(metadata.clone());
155    let request = Request::from_parts(
156        parts,
157        body.map_err(|_| Status::unavailable("request transport failed"))
158            .boxed_unsync(),
159    );
160    let admission = match core.prepare(&request, executor) {
161        Ok(admission) => admission,
162        Err(status) => return wire::failure(status, shape, true),
163    };
164    if admission.method.is_server_streaming() != (shape == wire::Shape::Stream) {
165        return admission.guard(
166            wire::failure(
167                Status::invalid_argument("content type does not match method shape"),
168                shape,
169                true,
170            ),
171            busy,
172        );
173    }
174    let maximum = wire::MAX_MESSAGE_BYTES + if shape == wire::Shape::Stream { 5 } else { 0 };
175    let (parts, body) = request.into_parts();
176    let bytes =
177        match tokio::time::timeout_at(admission.deadline, wire::collect(body, maximum)).await {
178            Ok(Ok(bytes)) => bytes,
179            Ok(Err(status)) => return admission.guard(wire::failure(status, shape, true), busy),
180            Err(_) => {
181                return admission.guard(
182                    wire::failure(
183                        Status::deadline_exceeded("RPC upload deadline exceeded"),
184                        shape,
185                        true,
186                    ),
187                    busy,
188                )
189            }
190        };
191    let bytes = match if shape == wire::Shape::Unary {
192        wire::unary_request(&bytes, request_gzip)
193    } else {
194        wire::streaming_request(&bytes, request_gzip)
195    } {
196        Ok(bytes) => bytes,
197        Err(status) => return admission.guard(wire::failure(status, shape, false), busy),
198    };
199    let response = core
200        .request(
201            Request::from_parts(parts, wire::body(bytes)),
202            connection,
203            registry,
204            &admission,
205        )
206        .await;
207    let response = match tokio::time::timeout_at(
208        admission.deadline,
209        wire::server_response(response, shape, metadata),
210    )
211    .await
212    {
213        Ok(Ok(response)) => response,
214        Ok(Err(status)) => wire::failure(status, shape, true),
215        Err(_) => wire::failure(
216            Status::deadline_exceeded("RPC response deadline exceeded"),
217            shape,
218            true,
219        ),
220    };
221    admission.guard(response, busy)
222}
223pub struct ConnectRpcServer;
224impl ConnectRpcServer {
225    pub async fn spawn(ctx: SpawnContext) -> Result<std::net::SocketAddr> {
226        let p = ctx.startup_params.as_ref();
227        let schema = p
228            .context("proto_schema required")?
229            .get_string("proto_schema")?;
230        let timeout = p
231            .map(|p| p.get_optional_u64("rpc_timeout_secs"))
232            .transpose()?
233            .flatten()
234            .unwrap_or(DEFAULT_RPC_TIMEOUT_SECS);
235        ensure!(
236            (1..=3600).contains(&timeout),
237            "rpc_timeout_secs must be 1..3600"
238        );
239        let pool = grpc::schema::load(&schema).await?;
240        ensure!(
241            pool.services().next().is_some(),
242            "schema must define services"
243        );
244        let core = HttpCore::new(
245            &ctx,
246            pool,
247            Arc::new(actions::ConnectRpcProtocol),
248            Duration::from_secs(timeout),
249        );
250        let listener =
251            crate::server::socket_helpers::create_reusable_tcp_listener(ctx.legacy_listen_addr())
252                .await?;
253        let address = listener.local_addr()?;
254        let registrar = ctx.state.clone();
255        let server_id = ctx.server_id;
256        let task = tokio::spawn(async move {
257            let limiter = crate::server::accept_bounded::ConnectionLimiter::new(MAX_CONNECTIONS);
258            loop {
259                let (stream, remote, permit) = match crate::server::accept_bounded::accept_bounded(
260                    &listener,
261                    &limiter,
262                    REFUSAL,
263                    "ConnectRPC",
264                    Some(&ctx.status_tx),
265                )
266                .await
267                {
268                    Ok(value) => value,
269                    Err(error) => {
270                        crate::console_error!(ctx.status_tx, "ConnectRPC accept failed: {error}");
271                        break;
272                    }
273                };
274                let id = crate::server::connection::ConnectionId::new(
275                    ctx.state.get_next_unified_id().await,
276                );
277                let now = crate::utils::clock::Instant::now();
278                ctx.state
279                    .add_connection_to_server(
280                        ctx.server_id,
281                        ConnectionState {
282                            id,
283                            remote_addr: remote,
284                            local_addr: address,
285                            bytes_sent: 0,
286                            bytes_received: 0,
287                            packets_sent: 0,
288                            packets_received: 0,
289                            last_activity: now,
290                            status: ConnectionStatus::Active,
291                            status_changed_at: now,
292                            protocol_info: ProtocolConnectionInfo::new(serde_json::json!({"transport":"http1","encoding":"proto","tls_verified":false})),
293                        },
294                    )
295                    .await;
296                let state = ctx.state.clone();
297                let owner = state.clone();
298                let core = core.clone();
299                let server = ctx.server_id;
300                owner.spawn_server_task(server, async move {
301                    let _permit = permit;
302                    if matches!(tokio::time::timeout(FIRST_BYTE_TIMEOUT, stream.peek(&mut [0u8;1])).await, Ok(Ok(n)) if n > 0) {
303                        let activity = Arc::new(crate::server::accept_bounded::ConnectionActivity::new());
304                        let executor = OwnedExecutor::default(); let _children = ConnectionTasks(executor.clone());
305                        let registry = Registry::default();
306                        let commands = crate::server::peer_support::register_peer_channel(&state, server, id.as_u32()).await;
307                        let _ = executor.spawn(core.clone().commands(commands, registry.clone(), id));
308                        let execute = executor.clone(); let active = activity.clone();
309                        let (done, ended) = tokio::sync::oneshot::channel();
310                        // Tracking the HTTP/1 owner in the same executor lets the core's hard
311                        // RPC timer cancel a body stalled behind a peer that never reads.
312                        let _ = executor.spawn(async move {
313                            let service = hyper::service::service_fn(move |request| {
314                                let core = core.clone(); let activity = active.clone(); let registry = registry.clone();
315                                let executor = execute.clone();
316                                async move { Ok::<_, Infallible>(self::request(request, core, id, activity, registry, executor).await) }
317                            });
318                            let _ = hyper::server::conn::http1::Builder::new().max_headers(64).max_buf_size(32768)
319                                .serve_connection(hyper_util::rt::TokioIo::new(stream), service).await;
320                            let _ = done.send(());
321                        });
322                        tokio::select! { _ = ended => {}, _ = crate::server::accept_bounded::watch_idle(activity, IDLE_TIMEOUT) => {} }
323                    }
324                    state.remove_peer_handle(server, id.as_u32()).await;
325                    state.remove_connection_from_server(server, id).await;
326                }).await;
327            }
328        });
329        registrar.register_server_task(server_id, task).await;
330        Ok(address)
331    }
332}
netget/server/connect_rpc/mod.rs

netget/server/connect_rpc/
mod.rs
