netget/server/http3/
mod.rs

1//! Real HTTP/3. The registered endpoint task owns every session and request future.
2pub mod actions;
3pub mod wire;
4use crate::llm::{action_helper::call_llm, ActionResult};
5use crate::logging::emit::Log;
6use crate::protocol::{Event, SpawnContext};
7use crate::server::connection::ConnectionId;
8use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
9use crate::utils::quic::*;
10use actions::{Http3Protocol, HTTP3_REQUEST_EVENT};
11use anyhow::{bail, ensure, Result};
12use bytes::{Buf, Bytes};
13use futures::{stream::FuturesUnordered, StreamExt};
14use serde_json::json;
15use std::{net::SocketAddr, sync::Arc, time::Duration};
16use wire::*;
17#[derive(Clone)]
18struct Settings {
19    handshake: Duration,
20    exchange: Duration,
21    idle: Duration,
22    connections: usize,
23    streams: usize,
24}
25impl Settings {
26    fn read(p: Option<&crate::protocol::StartupParams>) -> Result<Self> {
27        Ok(Self {
28            handshake: Duration::from_secs(bounded_parameter(p, "handshake_timeout_secs", 10, 60)?),
29            exchange: Duration::from_secs(bounded_parameter(p, "exchange_timeout_secs", 30, 300)?),
30            idle: Duration::from_secs(bounded_parameter(p, "idle_timeout_secs", 300, 3600)?),
31            connections: bounded_parameter(p, "max_connections", 64, 256)? as usize,
32            streams: bounded_parameter(p, "max_streams", 32, 32)? as usize,
33        })
34    }
35}
36pub struct Http3Server;
37impl Http3Server {
38    pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
39        let settings = Settings::read(ctx.startup_params.as_ref())?;
40        let cert = ctx
41            .startup_params
42            .as_ref()
43            .map(|p| p.get_optional_string("cert_path"))
44            .transpose()?
45            .flatten();
46        let key = ctx
47            .startup_params
48            .as_ref()
49            .map(|p| p.get_optional_string("key_path"))
50            .transpose()?
51            .flatten();
52        let _ = rustls::crypto::ring::default_provider().install_default();
53        let tls = match (cert, key) {
54            (Some(c), Some(k)) => {
55                crate::server::tls_cert_manager::load_tls_config_from_files(&c, &k)?
56            }
57            (None, None) => crate::server::tls_cert_manager::generate_default_tls_config()?,
58            _ => bail!("cert_path and key_path must be supplied together"),
59        };
60        let mut tls = (*tls).clone();
61        tls.alpn_protocols = vec![b"h3".to_vec()];
62        tls.max_early_data_size = 0;
63        let mut config = quinn::ServerConfig::with_crypto(Arc::new(
64            quinn::crypto::rustls::QuicServerConfig::try_from(tls)?,
65        ));
66        config.transport_config(transport(settings.idle, settings.streams as u32, 4));
67        config.migration(false);
68        let endpoint = EndpointOwner(EndpointGuard(quinn::Endpoint::server(
69            config,
70            ctx.legacy_listen_addr(),
71        )?));
72        let local = endpoint.local_addr()?;
73        Log::new(Some(&ctx.status_tx)).info(format!("HTTP3 listening on {local}"));
74        let state = ctx.state.clone();
75        let id = ctx.server_id;
76        let task = tokio::spawn(async move {
77            let mut sessions = FuturesUnordered::new();
78            loop {
79                tokio::select! {
80                    incoming=endpoint.accept()=> { let Some(incoming)=incoming else { break; }; if sessions.len()>=settings.connections { incoming.refuse(); continue; } sessions.push(Self::session(incoming,ctx.clone(),settings.clone(),local)); },
81                    _=sessions.next(),if !sessions.is_empty()=>{}
82                }
83            }
84        });
85        state.register_server_task(id, task).await;
86        Ok(local)
87    }
88    async fn session(
89        incoming: quinn::Incoming,
90        ctx: SpawnContext,
91        settings: Settings,
92        local: SocketAddr,
93    ) {
94        let connection = match tokio::time::timeout(settings.handshake, incoming).await {
95            Ok(Ok(c)) => ConnectionOwner(ConnectionGuard(c)),
96            _ => return,
97        };
98        let mut h3 = match h3::server::builder()
99            .send_grease(false)
100            .max_field_section_size(MAX_HEADERS as u64)
101            .build(h3_quinn::Connection::new(connection.clone()))
102            .await
103        {
104            Ok(c) => c,
105            Err(_) => return,
106        };
107        let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
108        let now = crate::utils::clock::Instant::now();
109        ctx.state
110            .add_connection_to_server(
111                ctx.server_id,
112                ConnectionState {
113                    id,
114                    remote_addr: connection.remote_address(),
115                    local_addr: local,
116                    bytes_sent: 0,
117                    bytes_received: 0,
118                    packets_sent: 0,
119                    packets_received: 0,
120                    last_activity: now,
121                    status: ConnectionStatus::Active,
122                    status_changed_at: now,
123                    protocol_info: ProtocolConnectionInfo::new(json!({"alpn":"h3"})),
124                },
125            )
126            .await;
127        let mut requests = FuturesUnordered::new();
128        loop {
129            tokio::select! {
130                resolver=h3.accept()=> { match resolver {
131                    Ok(Some(resolver))=> { if requests.len()>=settings.streams { connection.close(0x107u32.into(),b"request capacity"); break; } requests.push(Self::request(resolver,ctx.clone(),id,settings.exchange,connection.remote_address())); },
132                    _=>break
133                } },
134                _=requests.next(),if !requests.is_empty()=>{},
135                _=connection.closed()=>break,
136            }
137        }
138        drop(requests);
139        ctx.state
140            .close_connection_on_server(ctx.server_id, id)
141            .await;
142        let _ = ctx.status_tx.send("__UPDATE_UI__".into());
143    }
144    async fn request(
145        resolver: h3::server::RequestResolver<h3_quinn::Connection, Bytes>,
146        ctx: SpawnContext,
147        id: ConnectionId,
148        deadline: Duration,
149        peer: SocketAddr,
150    ) {
151        let end = tokio::time::Instant::now() + deadline;
152        let (request, stream) = match tokio::time::timeout_at(end, resolver.resolve_request()).await
153        {
154            Ok(Ok(r)) => r,
155            _ => return,
156        };
157        let mut guard = ServerStreamGuard {
158            stream,
159            done: false,
160        };
161        let exchange = async {
162            check_request_headers(request.headers())?;
163            let mut body = Vec::new();
164            while let Some(mut chunk) = guard.stream.recv_data().await? {
165                ensure!(
166                    chunk.remaining() <= MAX_BODY.saturating_sub(body.len()),
167                    "HTTP3 request body exceeds 8 MiB"
168                );
169                let count = chunk.remaining();
170                body.extend_from_slice(&chunk.copy_to_bytes(count));
171            }
172            let trailers = guard.stream.recv_trailers().await?.unwrap_or_default();
173            check_headers(&trailers)?;
174            validate_length(request.headers(), body.len())?;
175            let bytes = body.len();
176            let text = String::from_utf8(body)?;
177            ctx.state
178                .update_connection_stats(ctx.server_id, id, Some(bytes as u64), None, Some(1), None)
179                .await;
180            let event = Event::new(
181                &HTTP3_REQUEST_EVENT,
182                json!({
183                    "method":request.method().as_str(),
184                    "path":request.uri().path_and_query().map(|p|p.as_str()).unwrap_or("/"),
185                    "headers":request_header_json(request.headers())?, "body":text,
186                    "trailers":header_json(&trailers)?, "stream_id":guard.stream.id().index(), "peer_addr":peer.to_string(),
187                }),
188            );
189            Log::new(Some(&ctx.status_tx)).debug(format!(
190                "HTTP3 {} {}",
191                request.method(),
192                request.uri()
193            ));
194            Log::new(Some(&ctx.status_tx)).trace(format!("HTTP3 request: {}", event.data));
195            let log = Log::new(Some(&ctx.status_tx));
196            let response = match call_llm(
197                &ctx.llm_client,
198                &ctx.state,
199                ctx.server_id,
200                Some(id),
201                &event,
202                &Http3Protocol,
203            )
204            .await
205            {
206                Ok(result) => {
207                    for message in result.messages {
208                        log.info(message);
209                    }
210                    if !result.failures.is_empty() {
211                        log.error("HTTP3 decision=fail_closed_action_error; answering 500");
212                        json!({"status":500,"body":"Response could not be generated"})
213                    } else {
214                        let mut response = None;
215                        for action in result.protocol_results {
216                            match action {
217                                ActionResult::Custom { name, data } if name == "http3_response" => {
218                                    ensure!(response.is_none(), "Multiple final HTTP3 responses");
219                                    response = Some(data);
220                                }
221                                ActionResult::CloseConnection => {
222                                    log.info(
223                                        "HTTP3 decision=model_reject; request explicitly cancelled",
224                                    );
225                                    return Ok(false);
226                                }
227                                _ => {}
228                            }
229                        }
230                        if let Some(response) = response {
231                            let decision = if matches!(response["status"].as_u64(), Some(401 | 403))
232                            {
233                                "model_reject"
234                            } else {
235                                "model_answer"
236                            };
237                            log.info(format!(
238                                "HTTP3 decision={decision} status={}",
239                                response["status"]
240                            ));
241                            response
242                        } else {
243                            log.info("HTTP3 decision=model_silent; answering 500");
244                            json!({"status":500,"body":"No response provided"})
245                        }
246                    }
247                }
248                Err(error) => {
249                    let category = if crate::utils::WireFailure::classify(&error).is_overloaded() {
250                        "overloaded"
251                    } else {
252                        "unavailable"
253                    };
254                    log.error(format!("HTTP3 decision=fail_closed_llm_error category={category}; answering 503: {error}"));
255                    json!({"status":503,"body":"Service unavailable"})
256                }
257            };
258            let mut reply = http::Response::builder()
259                .status(response["status"].as_u64().unwrap_or(500) as u16)
260                .body(())?;
261            *reply.headers_mut() = parse_headers(&response["headers"])?;
262            let body = response["body"].as_str().unwrap_or("");
263            guard.stream.send_response(reply).await?;
264            let body = if request.method() == http::Method::HEAD {
265                ""
266            } else {
267                body
268            };
269            if !body.is_empty() {
270                guard
271                    .stream
272                    .send_data(Bytes::copy_from_slice(body.as_bytes()))
273                    .await?;
274            }
275            let trailers = parse_headers(&response["trailers"])?;
276            if !trailers.is_empty() {
277                guard.stream.send_trailers(trailers).await?;
278            }
279            guard.stream.finish().await?;
280            ctx.state
281                .update_connection_stats(
282                    ctx.server_id,
283                    id,
284                    None,
285                    Some(body.len() as u64),
286                    None,
287                    Some(1),
288                )
289                .await;
290            Ok::<_, anyhow::Error>(true)
291        };
292        match tokio::time::timeout_at(end, exchange).await {
293            Ok(Ok(true)) => guard.done = true,
294            Ok(Ok(false)) => {}
295            Ok(Err(error)) => Log::new(Some(&ctx.status_tx)).error(format!(
296                "HTTP3 decision=fail_closed_request_error; resetting request: {error}"
297            )),
298            Err(_) => Log::new(Some(&ctx.status_tx))
299                .error("HTTP3 decision=fail_closed_timeout; resetting request"),
300        }
301    }
302}
303struct ServerStreamGuard {
304    stream: h3::server::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
305    done: bool,
306}
307impl Drop for ServerStreamGuard {
308    fn drop(&mut self) {
309        if !self.done {
310            self.stream
311                .stop_sending(h3::error::Code::H3_REQUEST_CANCELLED);
312            self.stream
313                .stop_stream(h3::error::Code::H3_REQUEST_CANCELLED);
314        }
315    }
316}
netget/server/http3/mod.rs

netget/server/http3/
mod.rs
