origincloud_core/
transport.rs

1use std::time::Duration;
2
3use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};
4use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, HeaderValue, USER_AGENT};
5use serde::Serialize;
6use serde::de::DeserializeOwned;
7use serde_json::Value;
8
9use crate::consistency::is_malformed_token;
10use crate::error::ApiError;
11use crate::{Error, Service};
12
13/// The header carrying the consistency token, in responses and in
14/// requests.
15pub const CONSISTENCY_TOKEN_HEADER: &str = "Origin-Consistency-Token";
16
17/// The most of a response body that is read.
18const MAX_RESPONSE_BYTES: usize = 32 << 20;
19
20const FIRST_RETRY_DELAY: Duration = Duration::from_millis(200);
21const MAX_RETRY_DELAY: Duration = Duration::from_secs(10);
22
23/// The HTTP method of a [`Request`].
24pub use reqwest::Method;
25
26/// Query parameters, by their JSON paths.
27pub type Query = Vec<(String, String)>;
28
29/// What a method answering with a file rather than JSON returns
30/// (`google.api.HttpBody`): the file's bytes, as the server sent them, and
31/// their media type.
32#[derive(Clone, Debug, Default, PartialEq, Eq)]
33pub struct HttpBody {
34    /// The media type the response's `Content-Type` names, such as
35    /// `application/pdf`; empty when it names none.
36    pub content_type: String,
37    /// The file's bytes.
38    pub data: Vec<u8>,
39}
40
41/// The body of a 200, and its `Content-Type`.
42struct Answer {
43    content_type: String,
44    body: Vec<u8>,
45}
46
47/// One call of a service's method, as its HTTP binding sends it. The
48/// generated clients build it.
49#[derive(Clone, Debug)]
50pub struct Request {
51    method: Method,
52    path: String,
53    query: Query,
54    body: Option<Vec<u8>>,
55    idempotent: bool,
56    authenticated: bool,
57    /// A timeout answers the call rather than fails an attempt: it is
58    /// returned at once, not retried.
59    pub(crate) timeout_answers: bool,
60}
61
62impl Request {
63    /// A call of `method` on `path`, the escaped path from the root.
64    pub fn new(method: Method, path: impl Into<String>) -> Self {
65        Self {
66            method,
67            path: path.into(),
68            query: Query::new(),
69            body: None,
70            idempotent: false,
71            authenticated: true,
72            timeout_answers: false,
73        }
74    }
75
76    /// Sends `query` as the query parameters.
77    pub fn with_query(mut self, query: Query) -> Self {
78        self.query = query;
79        self
80    }
81
82    /// Sends `body`, JSON, as the body.
83    pub fn with_body(mut self, body: Vec<u8>) -> Self {
84        self.body = Some(body);
85        self
86    }
87
88    /// Marks the call as one whose sending twice has the effect of sending
89    /// it once: a read, or a mutation with a request id. Only such a call is
90    /// retried.
91    pub fn idempotent(mut self) -> Self {
92        self.idempotent = true;
93        self
94    }
95
96    /// Sends the call without the client's credentials, to a method that
97    /// authenticates no caller, such as the security token service's: it
98    /// then works with none.
99    ///
100    /// It is sent without the consistency token too, and the token its
101    /// answer carries, if any, isn't kept. In a browser it also carries no
102    /// `User-Agent`: only `Accept`, which a page may send to another
103    /// origin without asking first, and `Content-Type` with a body. So a
104    /// browser asks no CORS preflight of a read, and the call works
105    /// against an anonymous route whose CORS policy allows no request
106    /// headers. Natively it carries the `User-Agent` as every call does.
107    pub fn unauthenticated(mut self) -> Self {
108        self.authenticated = false;
109        self
110    }
111}
112
113impl Service {
114    /// The relative name `value`, the path field `field` of a call to this
115    /// service, stands for: `value` itself, or the relative name of a full
116    /// resource name it is. The generated clients call it, and send the
117    /// relative name in the path and in the body alike.
118    ///
119    /// `services` are the short codes of the services whose full names the
120    /// field takes: the service of the type the field refers to, such as
121    /// `resourcemanager` for a billing account's `parent`
122    /// (`organizations/*`), [`ANY_SERVICE`](crate::names::ANY_SERVICE) for
123    /// a field referring to any type, or none for this service's own. A full
124    /// name of one of them, such as
125    /// `//resourcemanager.origincloudapis.eu/organizations/o`, stands for its
126    /// relative name. A full name of another service, one with a regional or
127    /// zonal host or an API version, a URN, a URL or a malformed full name is
128    /// refused, naming the hosts the field takes; see
129    /// [`FullName`](crate::names::FullName).
130    pub fn relative_name(
131        &self,
132        field: &str,
133        value: &str,
134        services: &[&str],
135    ) -> Result<String, Error> {
136        crate::names::path_name(&self.shortcode, services, field, value)
137            .map(str::to_owned)
138            .map_err(Error::InvalidRequest)
139    }
140
141    /// `value`, the path field `field` of a call to this service, as the
142    /// path segments of the request: its [`relative_name`](Self::relative_name),
143    /// checked against `pattern` and escaped as [`path_value`] does.
144    pub fn path_value(
145        &self,
146        field: &str,
147        value: &str,
148        pattern: &str,
149        services: &[&str],
150    ) -> Result<String, Error> {
151        path_value(field, &self.relative_name(field, value, services)?, pattern)
152    }
153
154    /// Sends `request`, retrying it while that is safe and the answer
155    /// allows it, and reads the answer as `T`.
156    pub async fn call<T: DeserializeOwned>(&self, request: Request) -> Result<T, Error> {
157        let answer = self.send_with_retries(&request, "application/json").await?;
158        serde_json::from_slice(&answer.body).map_err(|e| {
159            Error::Decode(format!(
160                "reading the response to {} {}: {e}",
161                request.method, request.path
162            ))
163        })
164    }
165
166    /// Sends `request` as [`call`](Self::call) does, to a method answering
167    /// with a file rather than JSON, and returns the file as it came.
168    pub async fn call_http_body(&self, request: Request) -> Result<HttpBody, Error> {
169        let answer = self.send_with_retries(&request, "*/*").await?;
170        Ok(HttpBody {
171            content_type: answer.content_type,
172            data: answer.body,
173        })
174    }
175
176    /// Sends `request`, accepting `accept`, retrying it while that is safe
177    /// and the answer allows it.
178    ///
179    /// A call the API refuses because the consistency token it presented
180    /// isn't one (a token from another API, or one corrupted in storage) is
181    /// sent once more without it, whether it is retryable or not: the API
182    /// refuses that before doing anything. The token is dropped from the
183    /// store, so that later calls don't present it either.
184    async fn send_with_retries(&self, request: &Request, accept: &str) -> Result<Answer, Error> {
185        let tokens = &self.client.inner.tokens;
186        let mut attempt = 1;
187        let mut resent = false;
188        loop {
189            let presented = if request.authenticated {
190                tokens.get()
191            } else {
192                String::new()
193            };
194            let err = match self.send(request, accept, &presented).await {
195                Ok(answer) => return Ok(answer),
196                Err(err) => err,
197            };
198            if !resent && !presented.is_empty() && is_malformed_token(&err) {
199                tokens.remove(&presented);
200                resent = true;
201                continue;
202            }
203            if request.timeout_answers && is_timeout(&err) {
204                return Err(err);
205            }
206            if !request.idempotent || attempt >= self.max_attempts() {
207                return Err(err);
208            }
209            let Some(delay) = retry_delay(&err, attempt) else {
210                return Err(err);
211            };
212            crate::runtime::sleep(delay).await;
213            attempt += 1;
214        }
215    }
216
217    /// Makes one attempt, presenting the consistency token `presented` when
218    /// it isn't empty, and answers with the body of a 200.
219    async fn send(
220        &self,
221        request: &Request,
222        accept: &str,
223        presented: &str,
224    ) -> Result<Answer, Error> {
225        let base = self.base.as_ref().map_err(|e| Error::Config(e.clone()))?;
226        let mut url = base.clone();
227        url.set_path(&format!(
228            "{}{}",
229            base.path().trim_end_matches('/'),
230            request.path
231        ));
232        if request.query.is_empty() {
233            url.set_query(None);
234        } else {
235            url.query_pairs_mut().clear().extend_pairs(&request.query);
236        }
237        let inner = &self.client.inner;
238        let mut builder = inner.http.request(request.method.clone(), url);
239        builder = builder.header(ACCEPT, accept);
240        // In a browser an anonymous call is sent without it, as it would
241        // make the browser ask a CORS preflight.
242        if request.authenticated || cfg!(not(target_arch = "wasm32")) {
243            builder = builder.header(USER_AGENT, &inner.user_agent);
244        }
245        if request.authenticated {
246            let token = inner.credentials.token().await?;
247            builder = builder.header(AUTHORIZATION, format!("Bearer {token}"));
248        }
249        if let Some(timeout) = inner.timeout {
250            builder = builder.timeout(timeout);
251        }
252        if let Some(body) = &request.body {
253            builder = builder
254                .header(CONTENT_TYPE, "application/json")
255                .body(body.clone());
256        }
257        if !presented.is_empty() {
258            builder = builder.header(CONSISTENCY_TOKEN_HEADER, presented);
259        }
260        let response = builder.send().await.map_err(Error::Transport)?;
261        // A failed mutation may still have applied a command: its token
262        // comes back either way.
263        if let Some(answered) = response
264            .headers()
265            .get(CONSISTENCY_TOKEN_HEADER)
266            .and_then(|v: &HeaderValue| v.to_str().ok())
267            .filter(|v| !v.is_empty() && request.authenticated)
268        {
269            inner.tokens.replace(presented, answered.to_owned());
270        }
271        let status = response.status();
272        let content_type = response
273            .headers()
274            .get(CONTENT_TYPE)
275            .and_then(|v| v.to_str().ok())
276            .unwrap_or_default()
277            .to_owned();
278        let (answer, truncated) = read_capped(response, MAX_RESPONSE_BYTES)
279            .await
280            .map_err(Error::Transport)?;
281        if status != reqwest::StatusCode::OK {
282            return Err(ApiError::from_response(status.as_u16(), &answer).into());
283        }
284        // Cut short, a file would come back as another one.
285        if truncated {
286            return Err(Error::Decode(format!(
287                "the response to {} {} is larger than the {} MiB a response may be",
288                request.method,
289                request.path,
290                MAX_RESPONSE_BYTES >> 20
291            )));
292        }
293        Ok(Answer {
294            content_type,
295            body: answer,
296        })
297    }
298}
299
300/// A response's body, up to `max` bytes, and whether it was longer.
301/// Natively it is read chunk by chunk, and reading stops at the cap; in a
302/// browser, where reqwest's fetch has no `chunk()`, it is read whole and
303/// cut afterwards.
304pub(crate) async fn read_capped(
305    #[cfg_attr(target_arch = "wasm32", allow(unused_mut))] mut response: reqwest::Response,
306    max: usize,
307) -> Result<(Vec<u8>, bool), reqwest::Error> {
308    #[cfg(not(target_arch = "wasm32"))]
309    {
310        let mut body = Vec::new();
311        while let Some(chunk) = response.chunk().await? {
312            body.extend_from_slice(&chunk);
313            if body.len() > max {
314                body.truncate(max);
315                return Ok((body, true));
316            }
317        }
318        Ok((body, false))
319    }
320    #[cfg(target_arch = "wasm32")]
321    {
322        let mut body = Vec::from(response.bytes().await?);
323        let truncated = body.len() > max;
324        body.truncate(max);
325        Ok((body, truncated))
326    }
327}
328
329/// Whether `err` is an attempt that timed out.
330pub(crate) fn is_timeout(err: &Error) -> bool {
331    matches!(err, Error::Transport(e) if e.is_timeout())
332}
333
334/// Whether `err` may go away if the call is sent again, and how long to
335/// wait before attempt `attempt + 1`: what the answer asks for, or a
336/// jittered delay that doubles with each attempt, whichever is longer.
337fn retry_delay(err: &Error, attempt: u32) -> Option<Duration> {
338    let asked = match err {
339        Error::Transport(_) => Duration::ZERO,
340        Error::Api(e) if e.is_retryable() => e.retry_delay.unwrap_or_default(),
341        _ => return None,
342    };
343    let backoff = FIRST_RETRY_DELAY
344        .checked_mul(1 << attempt.saturating_sub(1).min(16))
345        .map_or(MAX_RETRY_DELAY, |b| b.min(MAX_RETRY_DELAY));
346    let half = backoff / 2;
347    let jitter = Duration::from_nanos(fastrand::u64(
348        0..=u64::try_from(half.as_nanos()).unwrap_or(u64::MAX),
349    ));
350    Some(asked.max(half + jitter))
351}
352
353/// What a path segment may hold unescaped: the unreserved characters, and
354/// the reserved ones a segment allows.
355const PATH_SEGMENT: &AsciiSet = &NON_ALPHANUMERIC
356    .remove(b'-')
357    .remove(b'_')
358    .remove(b'.')
359    .remove(b'~')
360    .remove(b'$')
361    .remove(b'&')
362    .remove(b'+')
363    .remove(b':')
364    .remove(b'=')
365    .remove(b'@');
366
367/// `value`, a relative resource name, as the path segments of a request,
368/// checking it has the form `pattern`, such as
369/// `projects/*/zones/*/volumes/*`, gives it. `field` names it in the
370/// error. [`Service::path_value`] also takes a full resource name of its
371/// service.
372pub fn path_value(field: &str, value: &str, pattern: &str) -> Result<String, Error> {
373    let got: Vec<&str> = value.split('/').collect();
374    let want: Vec<&str> = pattern.split('/').collect();
375    let valid = got.len() == want.len()
376        && got.iter().zip(&want).all(|(g, w)| match *w {
377            // A dot segment would move the request to another path, which
378            // the server would act on instead.
379            "*" => !g.is_empty() && *g != "." && *g != "..",
380            literal => *g == literal,
381        });
382    if !valid {
383        return Err(Error::InvalidRequest(format!(
384            "{field} must have the form {pattern}, got {value:?}"
385        )));
386    }
387    Ok(got
388        .iter()
389        .map(|segment| utf8_percent_encode(segment, PATH_SEGMENT).to_string())
390        .collect::<Vec<_>>()
391        .join("/"))
392}
393
394/// The fields of `message` as query parameters, by their JSON paths,
395/// leaving out the fields `skip` names and fields at their default value.
396/// A nested field is sent by its dotted path, a repeated one as one
397/// parameter per element. The generated clients send the fields their
398/// method's path and body don't carry this way.
399pub fn query<T: Serialize>(message: &T, skip: &[&str]) -> Result<Query, Error> {
400    let encoded = serde_json::to_value(message)
401        .map_err(|e| Error::InvalidRequest(format!("encoding the request: {e}")))?;
402    let mut query = Query::new();
403    if let Value::Object(fields) = encoded {
404        flatten(&mut query, "", &fields, skip)?;
405    }
406    query.sort_by(|a, b| a.0.cmp(&b.0));
407    Ok(query)
408}
409
410fn flatten(
411    query: &mut Query,
412    prefix: &str,
413    fields: &serde_json::Map<String, Value>,
414    skip: &[&str],
415) -> Result<(), Error> {
416    for (name, value) in fields {
417        let path = format!("{prefix}{name}");
418        if skip.contains(&path.as_str()) {
419            continue;
420        }
421        match value {
422            Value::Object(nested) => flatten(query, &format!("{path}."), nested, skip)?,
423            Value::Array(elements) => {
424                for element in elements {
425                    query.push((path.clone(), scalar(&path, element)?));
426                }
427            }
428            other => query.push((path.clone(), scalar(&path, other)?)),
429        }
430    }
431    Ok(())
432}
433
434fn scalar(path: &str, value: &Value) -> Result<String, Error> {
435    match value {
436        Value::String(s) => Ok(s.clone()),
437        Value::Number(n) => Ok(n.to_string()),
438        Value::Bool(b) => Ok(b.to_string()),
439        _ => Err(Error::InvalidRequest(format!(
440            "{path} can't be sent as a query parameter"
441        ))),
442    }
443}
444
445/// `message` as a JSON body. The generated clients call it.
446pub fn json_body<T: Serialize>(message: &T) -> Result<Vec<u8>, Error> {
447    serde_json::to_vec(message)
448        .map_err(|e| Error::InvalidRequest(format!("encoding the request: {e}")))
449}
450
451/// A new random request id, a UUID, for a mutation whose caller left its
452/// request id empty. The generated clients call it.
453pub fn new_request_id() -> String {
454    uuid::Uuid::new_v4().to_string()
455}
456
457#[cfg(test)]
458mod tests {
459    use super::*;
460
461    #[test]
462    fn a_path_value_is_checked_and_escaped() {
463        assert_eq!(
464            path_value(
465                "name",
466                "projects/p 1/zones/z/volumes/vol-1",
467                "projects/*/zones/*/volumes/*"
468            )
469            .unwrap(),
470            "projects/p%201/zones/z/volumes/vol-1"
471        );
472        assert_eq!(
473            path_value(
474                "name",
475                "projects/a b%/zones/z/volumes/v",
476                "projects/*/zones/*/volumes/*"
477            )
478            .unwrap(),
479            "projects/a%20b%25/zones/z/volumes/v"
480        );
481        for bad in [
482            "",
483            "projects/p/zones/z",
484            "projects/p/zones//volumes/v",
485            "projects/p/regions/z/volumes/v",
486            "projects/p/zones/z/volumes/v/x",
487            "projects/p/zones/z/volumes/..",
488            "projects/p/zones/z/volumes/.",
489            "projects/../zones/z/volumes/v",
490        ] {
491            let err = path_value("name", bad, "projects/*/zones/*/volumes/*").unwrap_err();
492            assert!(
493                err.to_string()
494                    .contains("name must have the form projects/*/zones/*/volumes/*"),
495                "{bad:?}: {err}"
496            );
497        }
498    }
499
500    #[test]
501    fn query_flattens_by_json_path() {
502        let message = serde_json::json!({
503            "parent": "projects/p/zones/z",
504            "pageSize": 10,
505            "filter": {"state": "AVAILABLE", "ids": ["a", "b"]},
506            "all": true,
507        });
508        let got = query(&message, &["parent"]).unwrap();
509        let encoded: Vec<String> = got.iter().map(|(k, v)| format!("{k}={v}")).collect();
510        assert_eq!(
511            encoded.join("&"),
512            "all=true&filter.ids=a&filter.ids=b&filter.state=AVAILABLE&pageSize=10"
513        );
514        let err = query(&serde_json::json!({"items": [{"a": 1}]}), &[]).unwrap_err();
515        assert!(
516            err.to_string()
517                .contains("items can't be sent as a query parameter")
518        );
519    }
520
521    #[test]
522    fn new_request_id_is_a_version_4_uuid() {
523        let mut seen = std::collections::HashSet::new();
524        for _ in 0..100 {
525            let id = new_request_id();
526            let parsed = uuid::Uuid::parse_str(&id).unwrap();
527            assert_eq!(parsed.get_version_num(), 4);
528            assert_eq!(id, id.to_lowercase());
529            assert!(seen.insert(id));
530        }
531    }
532
533    #[test]
534    fn the_retry_delay_grows_and_honours_the_answer() {
535        let unavailable = Error::from(ApiError::from_response(
536            503,
537            br#"{"error":{"status":"UNAVAILABLE","details":[{"@type":"type.googleapis.com/google.rpc.RetryInfo","retryDelay":"3s"}]}}"#,
538        ));
539        for attempt in 1..=10 {
540            let delay = retry_delay(&unavailable, attempt).unwrap();
541            assert!(
542                delay >= Duration::from_secs(3),
543                "attempt {attempt}: {delay:?}"
544            );
545            assert!(delay <= MAX_RETRY_DELAY, "attempt {attempt}: {delay:?}");
546        }
547        let proxy = Error::from(ApiError::from_response(502, b""));
548        let first = retry_delay(&proxy, 1).unwrap();
549        assert!(
550            first >= Duration::from_millis(100) && first <= Duration::from_millis(200),
551            "{first:?}"
552        );
553        let late = retry_delay(&proxy, 30).unwrap();
554        assert!(
555            late >= Duration::from_secs(5) && late <= MAX_RETRY_DELAY,
556            "{late:?}"
557        );
558        let refused = Error::from(ApiError::from_response(400, b""));
559        assert_eq!(retry_delay(&refused, 1), None);
560        assert_eq!(retry_delay(&Error::NoCredentials, 1), None);
561    }
562}
origincloud_core/transport.rs

origincloud_core/
transport.rs
