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

origincloud_core/
transport.rs
