objstore_r/
transport.rs

1// Author: spicyzhug@gmail.com
2// Copyright: lituus-io, all rights reserved
3//
4//! A minimal authorised-HTTP seam for Google REST APIs beyond Cloud Storage
5//! (Vertex AI, `BigQuery`), so dependants reuse this crate's TLS stack, token
6//! providers, retry policy and size caps instead of carrying their own HTTP
7//! client.
8//!
9//! The trait is dependency-free and always compiled; the reqwest-backed
10//! implementation is [`crate::http::Client`] (feature `http`). Futures are
11//! named associated types, never boxed, with no `Send`/`'static` bounds
12//! imposed here — the same rules as [`crate::ObjectStore`].
13//!
14//! Two ways to call:
15//!
16//! - [`Transport::send`]: one request, whole body back (bounded).
17//! - [`Transport::open`]: status now, body as a [`BodyStream`] of chunks
18//!   (server-sent events, long generations). **Dropping the stream cancels
19//!   the request**, which is how a caller stops paying for output it no
20//!   longer wants.
21
22pub use crate::key::encode_component;
23use crate::{Bytes, Result};
24use std::future::Future;
25
26/// HTTP method.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum Method {
29    /// `GET`
30    Get,
31    /// `POST`
32    Post,
33    /// `PUT`
34    Put,
35    /// `DELETE`
36    Delete,
37    /// `PATCH`
38    Patch,
39}
40
41/// When a failed attempt may be repeated.
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
43pub enum Retry {
44    /// Retry on transport errors and 408/429/5xx. Use for reads, and for
45    /// writes the server de-duplicates (e.g. a `BigQuery` `requestId`).
46    #[default]
47    Transient,
48    /// Retry only on 429 and on connect failures — cases where the server
49    /// provably did not start the work. Use for non-idempotent calls.
50    Throttled,
51    /// Never retry.
52    Never,
53}
54
55/// One request. Everything is borrowed except the body, which is a
56/// reference-counted [`Bytes`] so retries resend it without copying.
57#[derive(Debug, Clone)]
58pub struct Request<'a> {
59    /// Method.
60    pub method: Method,
61    /// Absolute `https://` URL (plain `http://` only to loopback).
62    pub url: &'a str,
63    /// Body (empty for none).
64    pub body: Bytes,
65    /// `content-type` of the body, when there is one.
66    pub content_type: Option<&'a str>,
67    /// Extra headers. `authorization` is set by the transport and must not
68    /// be passed here.
69    pub headers: &'a [(&'a str, &'a str)],
70    /// Retry policy.
71    pub retry: Retry,
72}
73
74impl<'a> Request<'a> {
75    /// `GET url`.
76    #[must_use]
77    pub fn get(url: &'a str) -> Self {
78        Self {
79            method: Method::Get,
80            url,
81            body: Bytes::new(),
82            content_type: None,
83            headers: &[],
84            retry: Retry::Transient,
85        }
86    }
87
88    /// `POST url` with a JSON body.
89    #[must_use]
90    pub fn post_json(url: &'a str, body: Bytes) -> Self {
91        Self {
92            method: Method::Post,
93            url,
94            body,
95            content_type: Some("application/json"),
96            headers: &[],
97            retry: Retry::Transient,
98        }
99    }
100
101    /// `PATCH url` with a JSON body. Retried only when the server provably
102    /// did not start the work ([`Retry::Throttled`]); pair it with an
103    /// `If-Match` header for optimistic concurrency.
104    #[must_use]
105    pub fn patch_json(url: &'a str, body: Bytes) -> Self {
106        Self {
107            method: Method::Patch,
108            url,
109            body,
110            content_type: Some("application/json"),
111            headers: &[],
112            retry: Retry::Throttled,
113        }
114    }
115
116    /// Replace the retry policy.
117    #[must_use]
118    pub fn retry(mut self, retry: Retry) -> Self {
119        self.retry = retry;
120        self
121    }
122
123    /// Replace the extra headers.
124    #[must_use]
125    pub fn headers(mut self, headers: &'a [(&'a str, &'a str)]) -> Self {
126        self.headers = headers;
127        self
128    }
129}
130
131/// A complete response.
132#[derive(Debug, Clone)]
133pub struct Response {
134    /// HTTP status.
135    pub status: u16,
136    /// Body (bounded by the transport's limit).
137    pub body: Bytes,
138}
139
140impl Response {
141    /// True for 2xx.
142    #[must_use]
143    pub fn is_success(&self) -> bool {
144        (200..300).contains(&self.status)
145    }
146
147    /// The error envelope of this (non-2xx) response.
148    #[must_use]
149    pub fn api_error(&self) -> ApiError {
150        ApiError::parse(self.status, &self.body)
151    }
152
153    /// `Ok(self)` for 2xx, else the status mapped to a typed [`crate::Error`]
154    /// (401/403 → `Auth`, 408/429/5xx → retriable `Unavailable`, others →
155    /// `Protocol`) carrying a short, control-character-free excerpt of the
156    /// server's message — never the request or a token.
157    pub fn error_for_status(self) -> Result<Self> {
158        if self.is_success() {
159            Ok(self)
160        } else {
161            Err(crate::net::status_error(self.status, &self.body))
162        }
163    }
164}
165
166/// A Google API error envelope (`{"error": {"code", "message", "errors":
167/// [{"reason"}], "status"}}`), reduced to what callers branch on. Text is
168/// control-character-free and bounded; a body that is not an envelope keeps
169/// a short excerpt as its message.
170#[derive(Debug, Clone, PartialEq, Eq)]
171pub struct ApiError {
172    /// HTTP status; for an error inside a 2xx body (a streamed error event,
173    /// parsed with status 0), the envelope's numeric `code`.
174    pub status: u16,
175    /// The first `reason` (for example `accessDenied`), or empty.
176    pub reason: String,
177    /// The server's message.
178    pub message: String,
179}
180
181impl ApiError {
182    /// Longest message kept.
183    pub const MAX_MESSAGE: usize = 500;
184
185    /// Parse an error body. Never fails: anything unparseable becomes the
186    /// message excerpt.
187    #[must_use]
188    pub fn parse(status: u16, body: &[u8]) -> Self {
189        let (mut message, mut reason, mut code) = (None, None, None);
190        json_members(body, |key, value, is_str| {
191            let slot = match (key, is_str) {
192                ("message", true) => &mut message,
193                ("reason", true) => &mut reason,
194                ("code", false) => {
195                    code = code.or_else(|| value.parse::<u16>().ok());
196                    return;
197                }
198                _ => return,
199            };
200            if slot.is_none() {
201                *slot = Some(clean(value));
202            }
203        });
204        let status = if status == 0 { code.unwrap_or(0) } else { status };
205        let message = message.unwrap_or_else(|| {
206            clean(&String::from_utf8_lossy(&body[..body.len().min(Self::MAX_MESSAGE)]))
207        });
208        Self { status, reason: reason.unwrap_or_default(), message }
209    }
210}
211
212fn clean(s: &str) -> String {
213    s.chars().filter(|c| !c.is_control()).take(ApiError::MAX_MESSAGE).collect()
214}
215
216/// Call `f(key, value, is_string)` for every scalar member of a JSON text
217/// (strings decoded, other scalars raw), at any depth, in document order;
218/// stops quietly at malformed input.
219fn json_members(b: &[u8], mut f: impl FnMut(&str, &str, bool)) {
220    let mut i = 0;
221    let mut key: Option<String> = None;
222    let mut after_colon = false;
223    while i < b.len() {
224        match b[i] {
225            b'"' => {
226                let Some((s, next)) = json_string(b, i + 1) else { return };
227                i = next;
228                if after_colon {
229                    if let Some(k) = key.take() {
230                        f(&k, &s, true);
231                    }
232                    after_colon = false;
233                } else {
234                    key = Some(s);
235                }
236            }
237            b':' => {
238                after_colon = key.is_some();
239                i += 1;
240            }
241            b',' | b'{' | b'}' | b'[' | b']' => {
242                key = None;
243                after_colon = false;
244                i += 1;
245            }
246            c if after_colon && !c.is_ascii_whitespace() => {
247                let end = b[i..]
248                    .iter()
249                    .position(|c| matches!(c, b',' | b'}' | b']') || c.is_ascii_whitespace())
250                    .map_or(b.len(), |n| i + n);
251                if let (Some(k), Ok(v)) = (key.take(), std::str::from_utf8(&b[i..end])) {
252                    f(&k, v, false);
253                }
254                after_colon = false;
255                i = end;
256            }
257            _ => i += 1,
258        }
259    }
260}
261
262/// Decode the JSON string starting after its opening quote at `i`; returns
263/// it and the index after its closing quote.
264fn json_string(b: &[u8], mut i: usize) -> Option<(String, usize)> {
265    let mut out = Vec::new();
266    while i < b.len() {
267        match b[i] {
268            b'"' => return Some((String::from_utf8_lossy(&out).into_owned(), i + 1)),
269            b'\\' => {
270                let e = *b.get(i + 1)?;
271                i += 2;
272                let c = match e {
273                    b'n' => '\n',
274                    b't' => '\t',
275                    b'r' => '\r',
276                    b'b' => '\u{8}',
277                    b'f' => '\u{c}',
278                    b'u' => {
279                        let hex = |at: usize| {
280                            std::str::from_utf8(b.get(at..at + 4)?)
281                                .ok()
282                                .and_then(|h| u32::from_str_radix(h, 16).ok())
283                        };
284                        let hi = hex(i)?;
285                        i += 4;
286                        if (0xD800..0xDC00).contains(&hi) && b.get(i..i + 2) == Some(b"\\u") {
287                            let lo = hex(i + 2)?;
288                            i += 6;
289                            char::from_u32(
290                                0x10000 + ((hi - 0xD800) << 10) + (lo.wrapping_sub(0xDC00) & 0x3FF),
291                            )
292                            .unwrap_or('\u{FFFD}')
293                        } else {
294                            char::from_u32(hi).unwrap_or('\u{FFFD}')
295                        }
296                    }
297                    other => char::from(other),
298                };
299                let mut tmp = [0; 4];
300                out.extend_from_slice(c.encode_utf8(&mut tmp).as_bytes());
301            }
302            c => {
303                out.push(c);
304                i += 1;
305            }
306        }
307    }
308    None
309}
310
311/// A response body delivered in chunks as they arrive.
312pub trait BodyStream {
313    /// Future of [`next_chunk`](BodyStream::next_chunk).
314    type Next<'a>: Future<Output = Result<Option<Bytes>>> + 'a
315    where
316        Self: 'a;
317
318    /// The next chunk, or `None` at the end of the body. The total is bounded
319    /// by the transport's limit.
320    fn next_chunk(&mut self) -> Self::Next<'_>;
321}
322
323/// An authorised HTTPS client.
324pub trait Transport {
325    /// Future of [`send`](Transport::send).
326    type Send<'a>: Future<Output = Result<Response>> + 'a
327    where
328        Self: 'a;
329    /// The chunked body of an [`open`](Transport::open)ed response.
330    type Body<'a>: BodyStream + 'a
331    where
332        Self: 'a;
333    /// Future of [`open`](Transport::open).
334    type Open<'a>: Future<Output = Result<(u16, Self::Body<'a>)>> + 'a
335    where
336        Self: 'a;
337
338    /// Send `req` and read the whole body. Non-2xx statuses are returned, not
339    /// raised (see [`Response::error_for_status`]); only transport failures
340    /// and limit violations are errors.
341    fn send<'a>(&'a self, req: Request<'a>) -> Self::Send<'a>;
342
343    /// Send `req` and return the status with a chunked body. Retries apply
344    /// until response headers arrive, never after the body has started.
345    fn open<'a>(&'a self, req: Request<'a>) -> Self::Open<'a>;
346}
347
348#[cfg(test)]
349mod tests {
350    use super::*;
351
352    #[test]
353    fn error_envelopes_parse_to_reason_and_message() {
354        // `U+` stands for a JSON `\u` escape.
355        let esc = |t: &str| t.replace("U+", "\\u");
356        let body =
357            esc(r#"{"error": {"code": 400, "message": "Unrecognized name: \"x\" at [1:8]U+00e9",
358            "errors": [{"message": "inner", "domain": "global", "reason": "invalidQuery"}],
359            "status": "INVALID_ARGUMENT"}}"#);
360        let e = ApiError::parse(400, body.as_bytes());
361        assert_eq!(e.reason, "invalidQuery");
362        assert_eq!(e.message, "Unrecognized name: \"x\" at [1:8]\u{e9}");
363        let raw = ApiError::parse(502, b"<html>bad\x1b[31m gateway</html>");
364        assert_eq!(
365            (raw.reason.as_str(), raw.message.as_str()),
366            ("", "<html>bad[31m gateway</html>")
367        );
368        let pair =
369            ApiError::parse(500, esc(r#"{"message": "U+d83dU+de00 x", "reason": 7}"#).as_bytes());
370        assert_eq!(pair.message, "\u{1F600} x");
371        for cut in 0..body.len() {
372            let _ = ApiError::parse(400, &body.as_bytes()[..cut]);
373        }
374        assert!(ApiError::parse(400, &[b'x'; 2000]).message.len() <= ApiError::MAX_MESSAGE);
375        // An error event inside a 2xx stream carries its own code; arrays wrap.
376        let ev = ApiError::parse(0, br#"[{"error":{"code":429,"message":"slow down"}}]"#);
377        assert_eq!((ev.status, ev.message.as_str()), (429, "slow down"));
378        assert_eq!(ApiError::parse(503, br#"{"code": 400}"#).status, 503);
379    }
380}
objstore_r/transport.rs

objstore_r/
transport.rs
