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

origincloud_core/
transport.rs
