netget/client/http3/
mod.rs

1//! Authenticated RFC 9114 client with one owned control driver and bounded request futures.
2//! The registered owner polls all query and handler futures; no detached tasks
3//! survive client removal and a parked handler never prevents command injection.
4pub mod actions;
5use crate::client::{command_support, llm_budget::call_llm_for_client};
6use crate::llm::actions::{
7    client_trait::{Client, ClientActionResult},
8    protocol_trait::Protocol,
9};
10use crate::logging::emit::Log;
11use crate::protocol::{ConnectContext, Event};
12use crate::server::http3::wire::*;
13use crate::state::{
14    client_handles::{ClientCommand, ClientSendOutcome},
15    AccessLogOwner, ClientStatus,
16};
17use crate::utils::quic::*;
18pub use actions::Http3ClientProtocol;
19use actions::{
20    HTTP3_CLIENT_CONNECTED_EVENT, HTTP3_CLIENT_ERROR_EVENT, HTTP3_CLIENT_RESPONSE_RECEIVED_EVENT,
21};
22use anyhow::{ensure, Context, Result};
23use bytes::{Buf, Bytes};
24type Sender = h3::client::SendRequest<h3_quinn::OpenStreams, Bytes>;
25use futures::{future::BoxFuture, stream::FuturesUnordered, FutureExt, StreamExt};
26use serde_json::{json, Value};
27use std::{net::SocketAddr, time::Duration};
28
29const MAX_FOLLOWUP_DEPTH: u8 = 4;
30type HandlerFuture = BoxFuture<'static, (u8, Result<Vec<Value>>)>;
31type QueryFuture = BoxFuture<'static, (u8, Event)>;
32
33pub struct Http3Client;
34impl Http3Client {
35    pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
36        let params = ctx.startup_params.as_ref();
37        let exchange =
38            Duration::from_secs(bounded_parameter(params, "exchange_timeout_secs", 30, 300)?);
39        let (endpoint, connection) =
40            crate::utils::quic::connect(&ctx.remote_addr, params, b"h3", 4).await?;
41        let endpoint = EndpointOwner(endpoint);
42        let connection = ConnectionOwner(connection);
43        let (driver, sender) = h3::client::builder()
44            .send_grease(false)
45            .max_field_section_size(MAX_HEADERS as u64)
46            .build(h3_quinn::Connection::new(connection.clone()))
47            .await?;
48        let local_addr = endpoint.local_addr()?;
49        let now = crate::utils::clock::Instant::now();
50        ctx.state
51            .with_client_mut(ctx.client_id, |client| {
52                client.connection = Some(crate::state::ClientConnectionState {
53                    id: ctx.client_id,
54                    remote_addr: ctx.remote_addr.clone(),
55                    connected_addr: Some(connection.remote_address()),
56                    local_addr: Some(local_addr),
57                    bytes_sent: 0,
58                    bytes_received: 0,
59                    packets_sent: 0,
60                    packets_received: 0,
61                    last_activity: now,
62                    status: ClientStatus::Connected,
63                    status_changed_at: now,
64                    protocol_info: crate::state::server::ProtocolConnectionInfo::new(
65                        json!({"alpn":"h3"}),
66                    ),
67                });
68            })
69            .await;
70        ctx.state
71            .update_client_status(ctx.client_id, ClientStatus::Connected)
72            .await;
73        let command_rx = command_support::register_command_channel(&ctx.state, ctx.client_id).await;
74        let state = ctx.state.clone();
75        let id = ctx.client_id;
76        let task = tokio::spawn(Self::run(
77            ctx, endpoint, connection, driver, sender, command_rx, exchange,
78        ));
79        state.register_client_task(id, task).await;
80        Ok(local_addr)
81    }
82
83    async fn run(
84        ctx: ConnectContext,
85        _endpoint: EndpointOwner,
86        connection: ConnectionOwner,
87        mut driver: h3::client::Connection<h3_quinn::Connection, Bytes>,
88        sender: Sender,
89        mut commands: tokio::sync::mpsc::Receiver<ClientCommand>,
90        exchange: Duration,
91    ) {
92        let mut handlers: FuturesUnordered<HandlerFuture> = FuturesUnordered::new();
93        let mut queries: FuturesUnordered<QueryFuture> = FuturesUnordered::new();
94        handlers.push(handle_event(
95            ctx.clone(),
96            Event::new(
97                &HTTP3_CLIENT_CONNECTED_EVENT,
98                json!({"base_url":format!("https://{}",ctx.remote_addr),"remote_addr":ctx.remote_addr}),
99            ),
100            0,
101        ));
102        let mut disconnect = false;
103        while !disconnect {
104            tokio::select! {
105                command = commands.recv() => {
106                    let Some(command) = command else { break; };
107                    disconnect = enqueue(&ctx,&sender,exchange,&mut queries,handlers.len(),command.action.clone(),Some(command),0).await;
108                }
109                Some((depth,event)) = queries.next(), if !queries.is_empty() => {
110                    // The final response still reaches the handler; only subsequent actions
111                    // are stopped at the depth boundary. A returned disconnect is honoured.
112                    handlers.push(handle_event(ctx.clone(),event,depth));
113                }
114                Some((depth,result)) = handlers.next(), if !handlers.is_empty() => {
115                    match result {
116                        Ok(actions) if actions.len() <= MAX_STREAMS => {
117                            for action in actions {
118                                if depth >= MAX_FOLLOWUP_DEPTH && action["type"] != "disconnect" {
119                                    if action["type"] == "send_http3_request" { Log::new(Some(&ctx.status_tx)).warn("HTTP3 follow-up limit reached"); }
120                                    continue;
121                                }
122                                if enqueue(&ctx,&sender,exchange,&mut queries,handlers.len(),action,None,depth).await { disconnect = true; break; }
123                            }
124                        }
125                        Ok(_) => Log::new(Some(&ctx.status_tx)).warn("HTTP3 handler returned more than 32 actions"),
126                        Err(e) => Log::new(Some(&ctx.status_tx)).warn(format!("HTTP3 event handler failed: {e}")),
127                    }
128                }
129                _ = std::future::poll_fn(|cx| driver.poll_close(cx)) => break,
130                _ = connection.closed() => break,
131            }
132        }
133        drop(queries);
134        drop(handlers);
135        connection.close(0x100u32.into(), b"client disconnected");
136        ctx.state
137            .with_client_mut(ctx.client_id, |client| {
138                if let Some(connection) = &mut client.connection {
139                    connection.status = ClientStatus::Disconnected;
140                    connection.status_changed_at = crate::utils::clock::Instant::now();
141                }
142            })
143            .await;
144        ctx.state.remove_client_handle(ctx.client_id).await;
145        ctx.state
146            .update_client_status(ctx.client_id, ClientStatus::Disconnected)
147            .await;
148        let _ = ctx.status_tx.send("__UPDATE_UI__".into());
149    }
150}
151
152fn handle_event(ctx: ConnectContext, event: Event, depth: u8) -> HandlerFuture {
153    async move {
154        let result = async {
155            let instruction = ctx
156                .state
157                .get_instruction_for_client(ctx.client_id)
158                .await
159                .unwrap_or_default();
160            let memory = ctx
161                .state
162                .get_memory_for_client(ctx.client_id)
163                .await
164                .unwrap_or_default();
165            let result = call_llm_for_client(
166                &ctx.llm_client,
167                &ctx.state,
168                ctx.client_id.to_string(),
169                &instruction,
170                &memory,
171                Some(&event),
172                &Http3ClientProtocol::new(),
173                &ctx.status_tx,
174            )
175            .await?;
176            if let Some(memory) = result.memory_updates {
177                ctx.state.set_memory_for_client(ctx.client_id, memory).await;
178            }
179            Ok(result.actions)
180        }
181        .await;
182        (depth, result)
183    }
184    .boxed()
185}
186
187async fn enqueue(
188    ctx: &ConnectContext,
189    sender: &Sender,
190    deadline: Duration,
191    queries: &mut FuturesUnordered<QueryFuture>,
192    handler_count: usize,
193    action: Value,
194    command: Option<ClientCommand>,
195    depth: u8,
196) -> bool {
197    let outcome = match Http3ClientProtocol::new().execute_action(action.clone()) {
198        Ok(ClientActionResult::Disconnect) => Ok(ClientSendOutcome::Disconnected),
199        Ok(ClientActionResult::WaitForMore | ClientActionResult::NoAction) => {
200            Ok(ClientSendOutcome::Executed {
201                detail: "Waiting for more queries".into(),
202            })
203        }
204        Ok(ClientActionResult::Custom { name, data }) if name == "http3_request" => {
205            // Reserve a handler slot for every accepted exchange, so a parked
206            // manual handler never causes a completed response event to be lost.
207            if queries.len() + handler_count >= MAX_STREAMS {
208                Err(anyhow::anyhow!(
209                    "HTTP3 client busy: 32 active exchanges or handlers"
210                ))
211            } else {
212                queries.push(query_future(
213                    ctx.clone(),
214                    sender.clone(),
215                    deadline,
216                    data,
217                    action,
218                    command,
219                    depth + 1,
220                ));
221                return false;
222            }
223        }
224        Ok(_) => Err(anyhow::anyhow!("Unsupported HTTP3 client action")),
225        Err(e) => Ok(ClientSendOutcome::Rejected {
226            error: e.to_string(),
227        }),
228    };
229    let disconnect = matches!(outcome, Ok(ClientSendOutcome::Disconnected));
230    record_action(ctx, &action, &outcome).await;
231    if let Some(command) = command {
232        command_support::reply(command, outcome);
233    }
234    disconnect
235}
236
237async fn record_action(ctx: &ConnectContext, action: &Value, outcome: &Result<ClientSendOutcome>) {
238    let result = match outcome {
239        Ok(v) => serde_json::to_value(v).unwrap_or(Value::Null),
240        Err(e) => json!({"error":e.to_string()}),
241    };
242    ctx.state
243        .record_access_log(
244            AccessLogOwner::Client(ctx.client_id.as_u32()),
245            Http3ClientProtocol::new().protocol_name(),
246            None,
247            "injected_action",
248            action.clone(),
249            vec![result],
250        )
251        .await;
252}
253
254fn query_future(
255    ctx: ConnectContext,
256    sender: Sender,
257    deadline: Duration,
258    data: Value,
259    action: Value,
260    command: Option<ClientCommand>,
261    depth: u8,
262) -> QueryFuture {
263    async move {
264        let result = async {
265            exchange(sender, &ctx.remote_addr, ctx.startup_params.as_ref(), data, deadline).await
266        }
267        .await;
268        let (outcome, event) = match result {
269            Ok(response) => {
270                ctx.state.with_client_mut(ctx.client_id, |client| {
271                    if let Some(connection) = &mut client.connection {
272                        connection.bytes_sent += action["body"].as_str().unwrap_or("").len() as u64;
273                        connection.bytes_received += response.body.len() as u64;
274                        connection.packets_sent += 1;
275                        connection.packets_received += 1;
276                        connection.last_activity = crate::utils::clock::Instant::now();
277                    }
278                }).await;
279                let event = Event::new(&HTTP3_CLIENT_RESPONSE_RECEIVED_EVENT, json!({
280                    "status_code":response.status_code,"headers":response.headers,"body":response.body,
281                    "trailers":response.trailers,"stream_id":response.stream_index,
282                }));
283                (Ok(ClientSendOutcome::Executed { detail: format!("http3_request {} {} -> {} ({} byte body)",
284                    action["method"].as_str().unwrap_or("GET"),action["path"].as_str().unwrap_or("/"),response.status_code,response.body.len()) }),event)
285            }
286            Err(e) => {
287                Log::new(Some(&ctx.status_tx)).warn(format!("HTTP3 stream failed: {e:#}"));
288                let event = Event::new(&HTTP3_CLIENT_ERROR_EVENT, json!({"error":format!("{e:#}")}));
289                (Err(e), event)
290            }
291        };
292        record_action(&ctx, &action, &outcome).await;
293        if let Some(command) = command {
294            command_support::reply(command, outcome);
295        }
296        (depth, event)
297    }
298    .boxed()
299}
300
301/// One semantic response. HTTP bodies exposed to handlers are UTF-8 text.
302#[derive(Debug)]
303pub struct Http3Exchange {
304    pub status_code: u16,
305    pub headers: serde_json::Map<String, Value>,
306    pub body: String,
307    pub trailers: serde_json::Map<String, Value>,
308    pub stream_index: u64,
309}
310/// Sends one request while the owner continues polling the control/QPACK driver.
311pub async fn exchange(
312    mut sender: Sender,
313    remote: &str,
314    params: Option<&crate::protocol::StartupParams>,
315    data: Value,
316    deadline: Duration,
317) -> Result<Http3Exchange> {
318    let method = data["method"].as_str().context("Missing method")?;
319    let path = data["path"].as_str().context("Missing path")?;
320    ensure!(
321        path.starts_with('/') && !path.starts_with("//"),
322        "HTTP3 path must be an origin-form path"
323    );
324    ensure!(path.len() <= MAX_HEADERS, "HTTP3 path too long");
325    let body = data["body"].as_str().unwrap_or("");
326    ensure!(body.len() <= MAX_BODY, "HTTP3 request body exceeds 8 MiB");
327    let end = tokio::time::Instant::now() + deadline;
328    let mut headers = params
329        .map(|p| p.get_optional_object("default_headers"))
330        .transpose()?
331        .flatten()
332        .cloned()
333        .unwrap_or_default();
334    headers = headers
335        .into_iter()
336        .map(|(k, v)| (k.to_ascii_lowercase(), v))
337        .collect();
338    if let Some(h) = data["headers"].as_object() {
339        for (key, value) in h {
340            headers.insert(key.to_ascii_lowercase(), value.clone());
341        }
342    }
343    if let Some(priority) = data["priority"].as_u64() {
344        ensure!(priority <= 7, "priority must be 0..7");
345        headers.insert("priority".into(), json!(format!("u={priority}")));
346    }
347    let mut request = http::Request::builder()
348        .method(method)
349        .uri(format!("https://{remote}{path}"))
350        .body(())?;
351    *request.headers_mut() = parse_request_headers(&Value::Object(headers))?;
352    check_field_section(
353        request.headers(),
354        &[
355            (":method", request.method().as_str()),
356            (":scheme", "https"),
357            (
358                ":authority",
359                request
360                    .uri()
361                    .authority()
362                    .context("Missing authority")?
363                    .as_str(),
364            ),
365            (":path", path),
366        ],
367    )?;
368    validate_length(request.headers(), body.len())?;
369    let stream = tokio::time::timeout_at(end, sender.send_request(request))
370        .await
371        .context("HTTP3 request deadline")??;
372    let mut stream = ClientStreamGuard {
373        stream,
374        done: false,
375    };
376    let result = tokio::time::timeout_at(end, async {
377        if !body.is_empty() {
378            stream
379                .stream
380                .send_data(Bytes::copy_from_slice(body.as_bytes()))
381                .await?;
382        }
383        let trailers = parse_headers(&data["trailers"])?;
384        if !trailers.is_empty() {
385            stream.stream.send_trailers(trailers).await?;
386        }
387        stream.stream.finish().await?;
388        let mut informational = 0;
389        let response = loop {
390            let response = stream.stream.recv_response().await?;
391            check_headers(response.headers())?;
392            if response.status().is_informational() {
393                ensure!(
394                    response.status().as_u16() != 101,
395                    "HTTP3 forbids switching protocols"
396                );
397                informational += 1;
398                ensure!(
399                    informational <= 16,
400                    "HTTP3 too many informational responses"
401                );
402                continue;
403            }
404            break response;
405        };
406        let mut bytes = Vec::new();
407        while let Some(mut chunk) = stream.stream.recv_data().await? {
408            ensure!(
409                chunk.remaining() <= MAX_BODY.saturating_sub(bytes.len()),
410                "HTTP3 response body exceeds 8 MiB"
411            );
412            let n = chunk.remaining();
413            bytes.extend_from_slice(&chunk.copy_to_bytes(n));
414        }
415        let trailers = stream.stream.recv_trailers().await?.unwrap_or_default();
416        check_headers(&trailers)?;
417        if method == "HEAD" || matches!(response.status().as_u16(), 204 | 205 | 304) {
418            ensure!(bytes.is_empty(), "HTTP3 response must not contain a body");
419        } else {
420            validate_length(response.headers(), bytes.len())?;
421        }
422        Ok(Http3Exchange {
423            status_code: response.status().as_u16(),
424            headers: header_json(response.headers())?,
425            body: String::from_utf8(bytes).context("HTTP3 body is not UTF-8 text")?,
426            trailers: header_json(&trailers)?,
427            stream_index: stream.stream.id().index(),
428        })
429    })
430    .await
431    .context("HTTP3 exchange deadline exceeded")?;
432    if result.is_ok() {
433        stream.done = true;
434    }
435    result
436}
437struct ClientStreamGuard {
438    stream: h3::client::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
439    done: bool,
440}
441impl Drop for ClientStreamGuard {
442    fn drop(&mut self) {
443        if !self.done {
444            self.stream
445                .stop_sending(h3::error::Code::H3_REQUEST_CANCELLED);
446            self.stream
447                .stop_stream(h3::error::Code::H3_REQUEST_CANCELLED);
448        }
449    }
450}
netget/client/http3/mod.rs

netget/client/http3/
mod.rs
