gemini_genai_rs/client/
http.rs

1//! HTTP client for Gemini REST APIs.
2//!
3//! Wraps `reqwest` with retry logic, telemetry, and typed errors.
4//! Feature-gated behind `http`.
5
6use std::time::Duration;
7
8use crate::telemetry;
9
10/// Configuration for the HTTP client.
11#[derive(Debug, Clone)]
12pub struct HttpConfig {
13    /// Request timeout.
14    pub timeout: Duration,
15    /// Maximum number of retries on transient errors (5xx, network).
16    pub max_retries: u32,
17    /// Base delay for exponential backoff between retries.
18    pub retry_base_delay: Duration,
19    /// Maximum delay between retries.
20    pub retry_max_delay: Duration,
21    /// User-Agent header value.
22    pub user_agent: String,
23}
24
25impl Default for HttpConfig {
26    fn default() -> Self {
27        Self {
28            timeout: Duration::from_secs(60),
29            max_retries: 3,
30            retry_base_delay: Duration::from_millis(500),
31            retry_max_delay: Duration::from_secs(30),
32            user_agent: format!("gemini-live/{}", env!("CARGO_PKG_VERSION")),
33        }
34    }
35}
36
37/// Errors from HTTP client operations.
38#[derive(Debug, thiserror::Error)]
39pub enum HttpError {
40    /// HTTP request failed.
41    #[error("HTTP request failed: {0}")]
42    Request(#[from] reqwest::Error),
43
44    /// Server returned an error status.
45    #[error("API error {status}: {message}")]
46    ApiError {
47        /// HTTP status code.
48        status: u16,
49        /// Error message from the API.
50        message: String,
51        /// Optional response body.
52        body: Option<serde_json::Value>,
53    },
54
55    /// Authentication error.
56    #[error("Auth error: {0}")]
57    Auth(#[from] crate::session::AuthError),
58
59    /// JSON deserialization error.
60    #[error("JSON parse error: {0}")]
61    Json(#[from] serde_json::Error),
62
63    /// All retries exhausted.
64    #[error("All {attempts} retries exhausted: {last_error}")]
65    RetriesExhausted {
66        /// Number of retry attempts made.
67        attempts: u32,
68        /// Error message from the last attempt.
69        last_error: String,
70    },
71}
72
73/// HTTP client wrapping reqwest with retry and telemetry.
74pub struct HttpClient {
75    inner: reqwest::Client,
76    config: HttpConfig,
77}
78
79impl HttpClient {
80    /// Create a new HTTP client with the given configuration.
81    pub fn new(config: HttpConfig) -> Self {
82        let inner = reqwest::Client::builder()
83            .timeout(config.timeout)
84            .user_agent(&config.user_agent)
85            .build()
86            .expect("Failed to build reqwest client");
87        Self { inner, config }
88    }
89
90    /// POST JSON to a URL and return the parsed response.
91    pub async fn post_json(
92        &self,
93        url: &str,
94        auth_headers: Vec<(String, String)>,
95        body: &impl serde::Serialize,
96    ) -> Result<serde_json::Value, HttpError> {
97        self.request_with_retry("POST", url, auth_headers, Some(body))
98            .await
99    }
100
101    /// PATCH JSON to a URL and return the parsed response.
102    pub async fn patch_json(
103        &self,
104        url: &str,
105        auth_headers: Vec<(String, String)>,
106        body: &impl serde::Serialize,
107    ) -> Result<serde_json::Value, HttpError> {
108        self.request_with_retry("PATCH", url, auth_headers, Some(body))
109            .await
110    }
111
112    /// PUT JSON to a URL and return the parsed response.
113    pub async fn put_json(
114        &self,
115        url: &str,
116        auth_headers: Vec<(String, String)>,
117        body: &impl serde::Serialize,
118    ) -> Result<serde_json::Value, HttpError> {
119        self.request_with_retry("PUT", url, auth_headers, Some(body))
120            .await
121    }
122
123    /// GET a URL and return the parsed response.
124    pub async fn get_json(
125        &self,
126        url: &str,
127        auth_headers: Vec<(String, String)>,
128    ) -> Result<serde_json::Value, HttpError> {
129        self.request_with_retry::<()>("GET", url, auth_headers, None)
130            .await
131    }
132
133    /// DELETE a URL and return the parsed response.
134    pub async fn delete(
135        &self,
136        url: &str,
137        auth_headers: Vec<(String, String)>,
138    ) -> Result<serde_json::Value, HttpError> {
139        self.request_with_retry::<()>("DELETE", url, auth_headers, None)
140            .await
141    }
142
143    /// POST JSON to a URL and stream the Server-Sent Events it answers with,
144    /// each event's `data` parsed as JSON.
145    ///
146    /// Transient failures are retried until the response starts, as for
147    /// [`post_json`](Self::post_json); after that the stream is not retried.
148    /// The client's timeout covers the whole stream.
149    pub async fn post_sse(
150        &self,
151        url: &str,
152        auth_headers: Vec<(String, String)>,
153        body: &impl serde::Serialize,
154    ) -> Result<
155        futures_util::stream::BoxStream<'static, Result<serde_json::Value, HttpError>>,
156        HttpError,
157    > {
158        use futures_util::StreamExt;
159
160        let response = self
161            .send_with_retry("POST", url, auth_headers, Some(body))
162            .await?;
163        let state = (
164            response,
165            SseDecoder::default(),
166            std::collections::VecDeque::<String>::new(),
167        );
168        let events = futures_util::stream::unfold(
169            state,
170            |(mut response, mut decoder, mut ready)| async move {
171                loop {
172                    if let Some(data) = ready.pop_front() {
173                        let parsed = serde_json::from_str::<serde_json::Value>(&data)
174                            .map_err(HttpError::from);
175                        return Some((parsed, (response, decoder, ready)));
176                    }
177                    match response.chunk().await {
178                        Ok(Some(bytes)) => ready.extend(decoder.push(&bytes)),
179                        Ok(None) => {
180                            ready.extend(decoder.finish());
181                            if ready.is_empty() {
182                                return None;
183                            }
184                        }
185                        Err(e) => {
186                            return Some((Err(HttpError::Request(e)), (response, decoder, ready)));
187                        }
188                    }
189                }
190            },
191        );
192        Ok(events.boxed())
193    }
194
195    /// Execute an HTTP request with exponential backoff retry on transient errors.
196    async fn request_with_retry<B: serde::Serialize>(
197        &self,
198        method: &str,
199        url: &str,
200        auth_headers: Vec<(String, String)>,
201        body: Option<&B>,
202    ) -> Result<serde_json::Value, HttpError> {
203        let response = self
204            .send_with_retry(method, url, auth_headers, body)
205            .await?;
206        let body = response.text().await?;
207        if body.is_empty() {
208            return Ok(serde_json::Value::Null);
209        }
210        Ok(serde_json::from_str(&body)?)
211    }
212
213    /// Send a request, retrying transient failures (5xx, 429, network) with
214    /// exponential backoff, and return the first successful response.
215    async fn send_with_retry<B: serde::Serialize>(
216        &self,
217        method: &str,
218        url: &str,
219        auth_headers: Vec<(String, String)>,
220        body: Option<&B>,
221    ) -> Result<reqwest::Response, HttpError> {
222        let mut last_error = String::new();
223
224        for attempt in 0..=self.config.max_retries {
225            if attempt > 0 {
226                let delay = self.backoff_delay(attempt);
227                telemetry::logging::log_http_retry(url, attempt, delay.as_millis() as u64);
228                tokio::time::sleep(delay).await;
229            }
230
231            telemetry::logging::log_http_request(method, url);
232            let _span = telemetry::spans::http_request_span(method, url);
233            let start = std::time::Instant::now();
234
235            match self.execute_request(method, url, &auth_headers, body).await {
236                Ok(response) => {
237                    let status = response.status();
238                    let duration_ms = start.elapsed().as_millis() as f64;
239                    telemetry::metrics::record_http_request(method, status.as_u16(), duration_ms);
240                    telemetry::logging::log_http_response(status.as_u16(), duration_ms);
241
242                    if status.is_success() {
243                        return Ok(response);
244                    }
245
246                    let status_code = status.as_u16();
247                    let body_text = response.text().await.unwrap_or_default();
248                    let body_json: Option<serde_json::Value> =
249                        serde_json::from_str(&body_text).ok();
250
251                    // Extract error message
252                    let message = body_json
253                        .as_ref()
254                        .and_then(|v| v.get("error"))
255                        .and_then(|v| v.get("message"))
256                        .and_then(|v| v.as_str())
257                        .unwrap_or(&body_text)
258                        .to_string();
259
260                    // Retry on 5xx and 429 (rate limit)
261                    if is_retryable_status(status_code) && attempt < self.config.max_retries {
262                        last_error = format!("HTTP {status_code}: {message}");
263                        continue;
264                    }
265
266                    return Err(HttpError::ApiError {
267                        status: status_code,
268                        message,
269                        body: body_json,
270                    });
271                }
272                Err(e) => {
273                    let duration_ms = start.elapsed().as_millis() as f64;
274                    telemetry::metrics::record_http_request(method, 0, duration_ms);
275
276                    if is_retryable_error(&e) && attempt < self.config.max_retries {
277                        last_error = e.to_string();
278                        continue;
279                    }
280                    return Err(HttpError::Request(e));
281                }
282            }
283        }
284
285        Err(HttpError::RetriesExhausted {
286            attempts: self.config.max_retries + 1,
287            last_error,
288        })
289    }
290
291    /// Execute a single HTTP request (no retry).
292    async fn execute_request<B: serde::Serialize>(
293        &self,
294        method: &str,
295        url: &str,
296        auth_headers: &[(String, String)],
297        body: Option<&B>,
298    ) -> Result<reqwest::Response, reqwest::Error> {
299        let mut builder = match method {
300            "POST" => self.inner.post(url),
301            "GET" => self.inner.get(url),
302            "DELETE" => self.inner.delete(url),
303            "PATCH" => self.inner.patch(url),
304            "PUT" => self.inner.put(url),
305            _ => self
306                .inner
307                .request(reqwest::Method::from_bytes(method.as_bytes()).unwrap(), url),
308        };
309
310        for (key, value) in auth_headers {
311            builder = builder.header(key, value);
312        }
313
314        if let Some(body) = body {
315            builder = builder.json(body);
316        }
317
318        builder.send().await
319    }
320
321    /// Calculate exponential backoff delay.
322    fn backoff_delay(&self, attempt: u32) -> Duration {
323        let delay = self.config.retry_base_delay * 2u32.saturating_pow(attempt.saturating_sub(1));
324        std::cmp::min(delay, self.config.retry_max_delay)
325    }
326}
327
328/// Whether an HTTP status code is retryable.
329/// Splits a Server-Sent Events byte stream into the `data` of each event.
330///
331/// Bytes are buffered until an event is complete (a blank line), so an event
332/// or a UTF-8 character split across network chunks is decoded whole. Lines
333/// other than `data:` (`event:`, `id:`, comments) are ignored; the lines of a
334/// multi-line `data` field are joined with `\n`.
335#[derive(Debug, Default)]
336pub struct SseDecoder {
337    buffer: Vec<u8>,
338}
339
340impl SseDecoder {
341    /// Feed received bytes; returns the `data` of every event they complete.
342    pub fn push(&mut self, bytes: &[u8]) -> Vec<String> {
343        self.buffer.extend_from_slice(bytes);
344        let mut events = Vec::new();
345        while let Some((end, separator)) = find_event_end(&self.buffer) {
346            let event: Vec<u8> = self.buffer.drain(..end + separator).collect();
347            if let Some(data) = event_data(&event[..end]) {
348                events.push(data);
349            }
350        }
351        events
352    }
353
354    /// The stream ended: return the last event if it was not terminated.
355    pub fn finish(&mut self) -> Vec<String> {
356        let rest = std::mem::take(&mut self.buffer);
357        event_data(&rest).into_iter().collect()
358    }
359}
360
361/// Where the first complete event ends, and the length of its separator.
362fn find_event_end(buffer: &[u8]) -> Option<(usize, usize)> {
363    (0..buffer.len()).find_map(|i| {
364        if buffer[i..].starts_with(b"\r\n\r\n") {
365            Some((i, 4))
366        } else if buffer[i..].starts_with(b"\n\n") {
367            Some((i, 2))
368        } else {
369            None
370        }
371    })
372}
373
374/// The joined `data:` lines of one event, if it has any.
375fn event_data(event: &[u8]) -> Option<String> {
376    let text = String::from_utf8_lossy(event);
377    let data: Vec<&str> = text
378        .lines()
379        .filter_map(|line| line.strip_prefix("data:"))
380        .map(|value| value.strip_prefix(' ').unwrap_or(value))
381        .collect();
382    (!data.is_empty()).then(|| data.join("\n"))
383}
384
385fn is_retryable_status(status: u16) -> bool {
386    status == 429 || (500..600).contains(&status)
387}
388
389/// Whether a reqwest error is retryable (network, timeout).
390fn is_retryable_error(e: &reqwest::Error) -> bool {
391    e.is_timeout() || e.is_connect() || e.is_request()
392}
393
394#[cfg(test)]
395mod tests {
396    use super::*;
397
398    #[test]
399    fn sse_events_survive_arbitrary_chunk_boundaries() {
400        let wire = "data: {\"a\":1}\n\n: comment\nevent: x\ndata: {\"b\":\"é\"}\r\n\r\ndata: [1,\ndata: 2]\n\n";
401        let bytes = wire.as_bytes();
402        // Every split point, including one inside the two-byte `é`.
403        for split in 0..=bytes.len() {
404            let mut decoder = SseDecoder::default();
405            let mut events = decoder.push(&bytes[..split]);
406            events.extend(decoder.push(&bytes[split..]));
407            events.extend(decoder.finish());
408            assert_eq!(
409                events,
410                ["{\"a\":1}", "{\"b\":\"é\"}", "[1,\n2]"],
411                "split at {split}"
412            );
413        }
414    }
415
416    /// Serve one canned HTTP response, written in the given pieces.
417    async fn serve_once(pieces: Vec<&'static str>) -> String {
418        use tokio::io::{AsyncReadExt, AsyncWriteExt};
419        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
420        let url = format!("http://{}/stream", listener.local_addr().unwrap());
421        tokio::spawn(async move {
422            let (mut socket, _) = listener.accept().await.unwrap();
423            let mut request = vec![0u8; 4096];
424            let _ = socket.read(&mut request).await;
425            for piece in pieces {
426                socket.write_all(piece.as_bytes()).await.unwrap();
427                socket.flush().await.unwrap();
428                tokio::time::sleep(Duration::from_millis(5)).await;
429            }
430        });
431        url
432    }
433
434    fn client() -> HttpClient {
435        HttpClient::new(HttpConfig {
436            max_retries: 0,
437            ..HttpConfig::default()
438        })
439    }
440
441    #[tokio::test]
442    async fn post_sse_yields_each_event_as_it_arrives() {
443        use futures_util::StreamExt;
444        let url = serve_once(vec![
445            "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\nconnection: close\r\n\r\n",
446            "data: {\"n\":1}\n",
447            "\ndata: {\"n\"",
448            ":2}\n\n",
449        ])
450        .await;
451        let events: Vec<_> = client()
452            .post_sse(&url, vec![], &serde_json::json!({}))
453            .await
454            .unwrap()
455            .collect()
456            .await;
457        let values: Vec<serde_json::Value> = events.into_iter().map(Result::unwrap).collect();
458        assert_eq!(
459            values,
460            [serde_json::json!({"n": 1}), serde_json::json!({"n": 2})]
461        );
462    }
463
464    #[tokio::test]
465    async fn post_sse_reports_an_error_status_before_streaming() {
466        let url = serve_once(vec![
467            "HTTP/1.1 400 Bad Request\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n",
468            "{\"error\":{\"message\":\"bad schema\"}}",
469        ])
470        .await;
471        match client()
472            .post_sse(&url, vec![], &serde_json::json!({}))
473            .await
474        {
475            Err(HttpError::ApiError {
476                status, message, ..
477            }) => {
478                assert_eq!(status, 400);
479                assert_eq!(message, "bad schema");
480            }
481            Err(other) => panic!("expected ApiError, got {other}"),
482            Ok(_) => panic!("expected ApiError, got a stream"),
483        }
484    }
485
486    #[test]
487    fn an_unterminated_last_event_is_kept() {
488        let mut decoder = SseDecoder::default();
489        assert!(decoder.push(b"data: {}").is_empty());
490        assert_eq!(decoder.finish(), ["{}"]);
491    }
492
493    #[test]
494    fn default_config() {
495        let config = HttpConfig::default();
496        assert_eq!(config.timeout, Duration::from_secs(60));
497        assert_eq!(config.max_retries, 3);
498        assert!(config.user_agent.starts_with("gemini-live/"));
499    }
500
501    #[test]
502    fn backoff_delay_calculation() {
503        let client = HttpClient::new(HttpConfig {
504            retry_base_delay: Duration::from_millis(100),
505            retry_max_delay: Duration::from_secs(5),
506            ..HttpConfig::default()
507        });
508        assert_eq!(client.backoff_delay(1), Duration::from_millis(100));
509        assert_eq!(client.backoff_delay(2), Duration::from_millis(200));
510        assert_eq!(client.backoff_delay(3), Duration::from_millis(400));
511    }
512
513    #[test]
514    fn backoff_delay_capped() {
515        let client = HttpClient::new(HttpConfig {
516            retry_base_delay: Duration::from_secs(1),
517            retry_max_delay: Duration::from_secs(5),
518            ..HttpConfig::default()
519        });
520        // 2^9 = 512 seconds, should be capped at 5 seconds
521        assert_eq!(client.backoff_delay(10), Duration::from_secs(5));
522    }
523
524    #[test]
525    fn retryable_status_codes() {
526        assert!(is_retryable_status(429));
527        assert!(is_retryable_status(500));
528        assert!(is_retryable_status(503));
529        assert!(is_retryable_status(599));
530        assert!(!is_retryable_status(400));
531        assert!(!is_retryable_status(404));
532        assert!(!is_retryable_status(200));
533    }
534}
gemini_genai_rs/client/http.rs

gemini_genai_rs/client/
http.rs
