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

origincloud_core/
transport.rs
