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    /// Sends `request`, retrying it while that is safe and the answer
106    /// allows it, and reads the answer as `T`.
107    pub async fn call<T: DeserializeOwned>(&self, request: Request) -> Result<T, Error> {
108        let answer = self.send_with_retries(&request, "application/json").await?;
109        serde_json::from_slice(&answer.body).map_err(|e| {
110            Error::Decode(format!(
111                "reading the response to {} {}: {e}",
112                request.method, request.path
113            ))
114        })
115    }
116
117    /// Sends `request` as [`call`](Self::call) does, to a method answering
118    /// with a file rather than JSON, and returns the file as it came.
119    pub async fn call_http_body(&self, request: Request) -> Result<HttpBody, Error> {
120        let answer = self.send_with_retries(&request, "*/*").await?;
121        Ok(HttpBody {
122            content_type: answer.content_type,
123            data: answer.body,
124        })
125    }
126
127    /// Sends `request`, accepting `accept`, retrying it while that is safe
128    /// and the answer allows it.
129    async fn send_with_retries(&self, request: &Request, accept: &str) -> Result<Answer, Error> {
130        let mut attempt = 1;
131        loop {
132            let err = match self.send(request, accept).await {
133                Ok(answer) => return Ok(answer),
134                Err(err) => err,
135            };
136            if request.timeout_answers && is_timeout(&err) {
137                return Err(err);
138            }
139            if !request.idempotent || attempt >= self.client.inner.max_attempts {
140                return Err(err);
141            }
142            let Some(delay) = retry_delay(&err, attempt) else {
143                return Err(err);
144            };
145            tokio::time::sleep(delay).await;
146            attempt += 1;
147        }
148    }
149
150    /// Makes one attempt, and answers with the body of a 200.
151    async fn send(&self, request: &Request, accept: &str) -> Result<Answer, Error> {
152        let base = self.base.as_ref().map_err(|e| Error::Config(e.clone()))?;
153        let mut url = base.clone();
154        url.set_path(&format!(
155            "{}{}",
156            base.path().trim_end_matches('/'),
157            request.path
158        ));
159        if request.query.is_empty() {
160            url.set_query(None);
161        } else {
162            url.query_pairs_mut().clear().extend_pairs(&request.query);
163        }
164        let inner = &self.client.inner;
165        let mut builder = inner.http.request(request.method.clone(), url);
166        if request.authenticated {
167            let token = inner.credentials.token().await?;
168            builder = builder.header(AUTHORIZATION, format!("Bearer {token}"));
169        }
170        builder = builder
171            .header(ACCEPT, accept)
172            .header(USER_AGENT, &inner.user_agent);
173        if let Some(timeout) = inner.timeout {
174            builder = builder.timeout(timeout);
175        }
176        if let Some(body) = &request.body {
177            builder = builder
178                .header(CONTENT_TYPE, "application/json")
179                .body(body.clone());
180        }
181        let held = self.client.consistency_token();
182        if !held.is_empty() {
183            builder = builder.header(CONSISTENCY_TOKEN_HEADER, held);
184        }
185        let mut response = builder.send().await.map_err(Error::Transport)?;
186        // A failed mutation may still have applied a command: its token
187        // comes back either way.
188        if let Some(answered) = response
189            .headers()
190            .get(CONSISTENCY_TOKEN_HEADER)
191            .and_then(|v: &HeaderValue| v.to_str().ok())
192            .filter(|v| !v.is_empty())
193        {
194            self.client.set_consistency_token(answered);
195        }
196        let status = response.status();
197        let content_type = response
198            .headers()
199            .get(CONTENT_TYPE)
200            .and_then(|v| v.to_str().ok())
201            .unwrap_or_default()
202            .to_owned();
203        let mut answer = Vec::new();
204        let mut truncated = false;
205        while let Some(chunk) = response.chunk().await.map_err(Error::Transport)? {
206            answer.extend_from_slice(&chunk);
207            if answer.len() > MAX_RESPONSE_BYTES {
208                answer.truncate(MAX_RESPONSE_BYTES);
209                truncated = true;
210                break;
211            }
212        }
213        if status != reqwest::StatusCode::OK {
214            return Err(ApiError::from_response(status.as_u16(), &answer).into());
215        }
216        // Cut short, a file would come back as another one.
217        if truncated {
218            return Err(Error::Decode(format!(
219                "the response to {} {} is larger than the {} MiB a response may be",
220                request.method,
221                request.path,
222                MAX_RESPONSE_BYTES >> 20
223            )));
224        }
225        Ok(Answer {
226            content_type,
227            body: answer,
228        })
229    }
230}
231
232/// Whether `err` is an attempt that timed out.
233pub(crate) fn is_timeout(err: &Error) -> bool {
234    matches!(err, Error::Transport(e) if e.is_timeout())
235}
236
237/// Whether `err` may go away if the call is sent again, and how long to
238/// wait before attempt `attempt + 1`: what the answer asks for, or a
239/// jittered delay that doubles with each attempt, whichever is longer.
240fn retry_delay(err: &Error, attempt: u32) -> Option<Duration> {
241    let asked = match err {
242        Error::Transport(_) => Duration::ZERO,
243        Error::Api(e) if e.is_retryable() => e.retry_delay.unwrap_or_default(),
244        _ => return None,
245    };
246    let backoff = FIRST_RETRY_DELAY
247        .checked_mul(1 << attempt.saturating_sub(1).min(16))
248        .map_or(MAX_RETRY_DELAY, |b| b.min(MAX_RETRY_DELAY));
249    let half = backoff / 2;
250    let jitter = Duration::from_nanos(fastrand::u64(
251        0..=u64::try_from(half.as_nanos()).unwrap_or(u64::MAX),
252    ));
253    Some(asked.max(half + jitter))
254}
255
256/// What a path segment may hold unescaped: the unreserved characters, and
257/// the reserved ones a segment allows.
258const PATH_SEGMENT: &AsciiSet = &NON_ALPHANUMERIC
259    .remove(b'-')
260    .remove(b'_')
261    .remove(b'.')
262    .remove(b'~')
263    .remove(b'$')
264    .remove(b'&')
265    .remove(b'+')
266    .remove(b':')
267    .remove(b'=')
268    .remove(b'@');
269
270/// `value` as the path segments of a request, checking it has the form
271/// `pattern`, such as `projects/*/zones/*/volumes/*`, gives it. `field`
272/// names it in the error. The generated clients call it.
273pub fn path_value(field: &str, value: &str, pattern: &str) -> Result<String, Error> {
274    let got: Vec<&str> = value.split('/').collect();
275    let want: Vec<&str> = pattern.split('/').collect();
276    let valid = got.len() == want.len()
277        && got.iter().zip(&want).all(|(g, w)| match *w {
278            // A dot segment would move the request to another path, which
279            // the server would act on instead.
280            "*" => !g.is_empty() && *g != "." && *g != "..",
281            literal => *g == literal,
282        });
283    if !valid {
284        return Err(Error::InvalidRequest(format!(
285            "{field} must have the form {pattern}, got {value:?}"
286        )));
287    }
288    Ok(got
289        .iter()
290        .map(|segment| utf8_percent_encode(segment, PATH_SEGMENT).to_string())
291        .collect::<Vec<_>>()
292        .join("/"))
293}
294
295/// The fields of `message` as query parameters, by their JSON paths,
296/// leaving out the fields `skip` names and fields at their default value.
297/// A nested field is sent by its dotted path, a repeated one as one
298/// parameter per element. The generated clients send the fields their
299/// method's path and body don't carry this way.
300pub fn query<T: Serialize>(message: &T, skip: &[&str]) -> Result<Query, Error> {
301    let encoded = serde_json::to_value(message)
302        .map_err(|e| Error::InvalidRequest(format!("encoding the request: {e}")))?;
303    let mut query = Query::new();
304    if let Value::Object(fields) = encoded {
305        flatten(&mut query, "", &fields, skip)?;
306    }
307    query.sort_by(|a, b| a.0.cmp(&b.0));
308    Ok(query)
309}
310
311fn flatten(
312    query: &mut Query,
313    prefix: &str,
314    fields: &serde_json::Map<String, Value>,
315    skip: &[&str],
316) -> Result<(), Error> {
317    for (name, value) in fields {
318        let path = format!("{prefix}{name}");
319        if skip.contains(&path.as_str()) {
320            continue;
321        }
322        match value {
323            Value::Object(nested) => flatten(query, &format!("{path}."), nested, skip)?,
324            Value::Array(elements) => {
325                for element in elements {
326                    query.push((path.clone(), scalar(&path, element)?));
327                }
328            }
329            other => query.push((path.clone(), scalar(&path, other)?)),
330        }
331    }
332    Ok(())
333}
334
335fn scalar(path: &str, value: &Value) -> Result<String, Error> {
336    match value {
337        Value::String(s) => Ok(s.clone()),
338        Value::Number(n) => Ok(n.to_string()),
339        Value::Bool(b) => Ok(b.to_string()),
340        _ => Err(Error::InvalidRequest(format!(
341            "{path} can't be sent as a query parameter"
342        ))),
343    }
344}
345
346/// `message` as a JSON body. The generated clients call it.
347pub fn json_body<T: Serialize>(message: &T) -> Result<Vec<u8>, Error> {
348    serde_json::to_vec(message)
349        .map_err(|e| Error::InvalidRequest(format!("encoding the request: {e}")))
350}
351
352/// A new random request id, a UUID, for a mutation whose caller left its
353/// request id empty. The generated clients call it.
354pub fn new_request_id() -> String {
355    uuid::Uuid::new_v4().to_string()
356}
357
358#[cfg(test)]
359mod tests {
360    use super::*;
361
362    #[test]
363    fn a_path_value_is_checked_and_escaped() {
364        assert_eq!(
365            path_value(
366                "name",
367                "projects/p 1/zones/z/volumes/vol-1",
368                "projects/*/zones/*/volumes/*"
369            )
370            .unwrap(),
371            "projects/p%201/zones/z/volumes/vol-1"
372        );
373        assert_eq!(
374            path_value(
375                "name",
376                "projects/a b%/zones/z/volumes/v",
377                "projects/*/zones/*/volumes/*"
378            )
379            .unwrap(),
380            "projects/a%20b%25/zones/z/volumes/v"
381        );
382        for bad in [
383            "",
384            "projects/p/zones/z",
385            "projects/p/zones//volumes/v",
386            "projects/p/regions/z/volumes/v",
387            "projects/p/zones/z/volumes/v/x",
388            "projects/p/zones/z/volumes/..",
389            "projects/p/zones/z/volumes/.",
390            "projects/../zones/z/volumes/v",
391        ] {
392            let err = path_value("name", bad, "projects/*/zones/*/volumes/*").unwrap_err();
393            assert!(
394                err.to_string()
395                    .contains("name must have the form projects/*/zones/*/volumes/*"),
396                "{bad:?}: {err}"
397            );
398        }
399    }
400
401    #[test]
402    fn query_flattens_by_json_path() {
403        let message = serde_json::json!({
404            "parent": "projects/p/zones/z",
405            "pageSize": 10,
406            "filter": {"state": "AVAILABLE", "ids": ["a", "b"]},
407            "all": true,
408        });
409        let got = query(&message, &["parent"]).unwrap();
410        let encoded: Vec<String> = got.iter().map(|(k, v)| format!("{k}={v}")).collect();
411        assert_eq!(
412            encoded.join("&"),
413            "all=true&filter.ids=a&filter.ids=b&filter.state=AVAILABLE&pageSize=10"
414        );
415        let err = query(&serde_json::json!({"items": [{"a": 1}]}), &[]).unwrap_err();
416        assert!(
417            err.to_string()
418                .contains("items can't be sent as a query parameter")
419        );
420    }
421
422    #[test]
423    fn new_request_id_is_a_version_4_uuid() {
424        let mut seen = std::collections::HashSet::new();
425        for _ in 0..100 {
426            let id = new_request_id();
427            let parsed = uuid::Uuid::parse_str(&id).unwrap();
428            assert_eq!(parsed.get_version_num(), 4);
429            assert_eq!(id, id.to_lowercase());
430            assert!(seen.insert(id));
431        }
432    }
433
434    #[test]
435    fn the_retry_delay_grows_and_honours_the_answer() {
436        let unavailable = Error::from(ApiError::from_response(
437            503,
438            br#"{"error":{"status":"UNAVAILABLE","details":[{"@type":"type.googleapis.com/google.rpc.RetryInfo","retryDelay":"3s"}]}}"#,
439        ));
440        for attempt in 1..=10 {
441            let delay = retry_delay(&unavailable, attempt).unwrap();
442            assert!(
443                delay >= Duration::from_secs(3),
444                "attempt {attempt}: {delay:?}"
445            );
446            assert!(delay <= MAX_RETRY_DELAY, "attempt {attempt}: {delay:?}");
447        }
448        let proxy = Error::from(ApiError::from_response(502, b""));
449        let first = retry_delay(&proxy, 1).unwrap();
450        assert!(
451            first >= Duration::from_millis(100) && first <= Duration::from_millis(200),
452            "{first:?}"
453        );
454        let late = retry_delay(&proxy, 30).unwrap();
455        assert!(
456            late >= Duration::from_secs(5) && late <= MAX_RETRY_DELAY,
457            "{late:?}"
458        );
459        let refused = Error::from(ApiError::from_response(400, b""));
460        assert_eq!(retry_delay(&refused, 1), None);
461        assert_eq!(retry_delay(&Error::NoCredentials, 1), None);
462    }
463}
origincloud_core/transport.rs

origincloud_core/
transport.rs
