ocpi_kit/client/
http.rs

1//! The request executor: auth, headers, retries, envelope decoding, tracing.
2
3use http::{HeaderMap, HeaderValue, Method};
4use serde::Serialize;
5use serde::de::DeserializeOwned;
6use tracing::Instrument as _;
7
8use crate::ModuleId;
9use crate::transport::headers::{APPLICATION_JSON, AUTHORIZATION};
10use crate::transport::{
11    CredentialsToken, OcpiError, OcpiResponse, Page, PageMeta, Quirks, RequestIds, RoutingHeaders,
12};
13use crate::types::{Url, UrlPolicy, Validate};
14
15use super::{ClientConfig, RetryPolicy};
16
17/// One outgoing OCPI request, before it is sent.
18#[derive(Debug)]
19pub struct OcpiRequest {
20    /// The HTTP method.
21    pub method: Method,
22    /// The absolute URL to call.
23    pub url: Url,
24    /// The module the request addresses, which decides whether routing headers apply.
25    pub module: ModuleId,
26    /// The routing headers, when the module is a functional one.
27    pub routing: Option<RoutingHeaders>,
28    /// The request and correlation IDs.
29    pub ids: RequestIds,
30    /// The JSON body, already serialised.
31    pub body: Option<Vec<u8>>,
32}
33
34impl OcpiRequest {
35    /// A request with freshly generated IDs and no body.
36    #[must_use]
37    pub fn new(method: Method, url: Url, module: ModuleId) -> Self {
38        Self { method, url, module, routing: None, ids: RequestIds::generate(), body: None }
39    }
40
41    /// Attaches routing headers, which are dropped for a configuration module.
42    ///
43    /// > *routing headers SHALL NOT be used with these modules*
44    #[must_use]
45    pub fn routed(mut self, routing: RoutingHeaders) -> Self {
46        if self.module.is_functional() {
47            self.routing = Some(routing);
48        }
49        self
50    }
51
52    /// Attaches the request and correlation IDs.
53    #[must_use]
54    pub fn with_ids(mut self, ids: RequestIds) -> Self {
55        self.ids = ids;
56        self
57    }
58
59    /// Serialises `body` as the request body.
60    ///
61    /// # Errors
62    ///
63    /// Returns [`OcpiError::Decode`] if the value cannot be serialised.
64    pub fn with_body<T: Serialize>(mut self, body: &T) -> Result<Self, OcpiError> {
65        self.body = Some(
66            serde_json::to_vec(body)
67                .map_err(|e| OcpiError::Decode { path: "/".to_owned(), message: e.to_string() })?,
68        );
69        Ok(self)
70    }
71
72    /// Whether the specification permits retrying this request automatically.
73    ///
74    /// > *OCPI messages SHOULD NOT be queued. When a client does a POST, PUT or PATCH request and
75    /// > that request fails or times out, the client should not queue the message and retry the
76    /// > same message again later.*
77    ///
78    /// Only `GET` is retryable.
79    #[must_use]
80    pub fn is_retryable(&self) -> bool {
81        self.method == Method::GET
82    }
83
84    fn header_map(&self, token: &CredentialsToken, quirks: &Quirks) -> HeaderMap {
85        let mut headers = HeaderMap::new();
86        let auth = if quirks.send_unencoded_token {
87            token.to_header_value_unencoded()
88        } else {
89            token.to_header_value()
90        };
91        if let Ok(value) = HeaderValue::from_str(&auth) {
92            headers.insert(AUTHORIZATION, value);
93        }
94        self.ids.write_to(&mut headers);
95        if let Some(routing) = &self.routing
96            && !quirks.omit_routing_headers
97        {
98            routing.write_to(&mut headers);
99        }
100        if self.body.is_some() {
101            headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static(APPLICATION_JSON));
102        }
103        headers
104    }
105}
106
107/// Sends OCPI requests and decodes the envelope.
108#[derive(Clone, Debug)]
109pub struct Transport {
110    http: reqwest::Client,
111    config: ClientConfig,
112}
113
114impl Transport {
115    /// Wraps a `reqwest` client.
116    #[must_use]
117    pub fn new(http: reqwest::Client, config: ClientConfig) -> Self {
118        Self { http, config }
119    }
120
121    /// The URL policy every outgoing request is checked against.
122    #[must_use]
123    pub const fn url_policy(&self) -> &UrlPolicy {
124        &self.config.url_policy
125    }
126
127    /// The configuration in use.
128    #[must_use]
129    pub const fn config(&self) -> &ClientConfig {
130        &self.config
131    }
132
133    /// Sends a request and decodes the response envelope into `T`.
134    ///
135    /// # Errors
136    ///
137    /// Returns [`OcpiError`] for a refused URL, a transport failure, a body that is not the
138    /// expected shape, or a non-success OCPI status code.
139    pub async fn send<T: DeserializeOwned>(
140        &self,
141        request: &OcpiRequest,
142        token: &CredentialsToken,
143        quirks: &Quirks,
144    ) -> Result<T, OcpiError> {
145        let (response, _) = self.send_with_headers::<T>(request, token, quirks).await?;
146        response.into_result()
147    }
148
149    /// Sends a request and returns one page of a list endpoint.
150    ///
151    /// # Errors
152    ///
153    /// As [`Transport::send`].
154    pub async fn send_page<T: DeserializeOwned>(
155        &self,
156        request: &OcpiRequest,
157        token: &CredentialsToken,
158        quirks: &Quirks,
159    ) -> Result<Page<T>, OcpiError> {
160        let (response, headers) = self.send_with_headers::<Vec<T>>(request, token, quirks).await?;
161        let meta = PageMeta::from_headers(&headers);
162        Ok(Page { items: response.into_list()?, meta })
163    }
164
165    /// Sends a request and returns the envelope together with the response headers.
166    ///
167    /// # Errors
168    ///
169    /// As [`Transport::send`], except that a non-success status code is returned in the envelope
170    /// rather than as an error.
171    pub async fn send_with_headers<T: DeserializeOwned>(
172        &self,
173        request: &OcpiRequest,
174        token: &CredentialsToken,
175        quirks: &Quirks,
176    ) -> Result<(OcpiResponse<T>, HeaderMap), OcpiError> {
177        self.config.url_policy.check(&request.url).map_err(|e| OcpiError::UrlRefused {
178            url: request.url.as_str().to_owned(),
179            reason: e.to_string(),
180        })?;
181
182        let retries = if request.is_retryable() { self.config.retry.max_attempts } else { 1 };
183        let mut attempt = 0u32;
184        loop {
185            attempt += 1;
186            match self.attempt(request, token, quirks).await {
187                Ok(result) => return Ok(result),
188                Err(error) if attempt < retries && error.is_transient() => {
189                    #[cfg(feature = "client")]
190                    tracing::debug!(
191                        attempt,
192                        %error,
193                        url = request.url.as_str(),
194                        "retrying a GET after a transient failure",
195                    );
196                    tokio::time::sleep(
197                        self.config.retry.delay_for(attempt, RetryPolicy::seed_from(&request.ids.request_id)),
198                    )
199                    .await;
200                }
201                Err(error) => return Err(error),
202            }
203        }
204    }
205
206    async fn attempt<T: DeserializeOwned>(
207        &self,
208        request: &OcpiRequest,
209        token: &CredentialsToken,
210        quirks: &Quirks,
211    ) -> Result<(OcpiResponse<T>, HeaderMap), OcpiError> {
212        // The span is attached with `Instrument` rather than entered with a guard: an
213        // `Entered` guard held across an `.await` stays entered while the task is parked, so
214        // whatever the executor polls next inherits this request's span. `Instrument` enters
215        // and exits around each poll, which is the only correct form in an async fn.
216        let span = tracing::info_span!(
217            "ocpi.request",
218            otel.kind = "client",
219            http.request.method = %request.method,
220            url.full = request.url.as_str(),
221            ocpi.module = %request.module,
222            ocpi.request_id = request.ids.request_id.as_str(),
223            ocpi.correlation_id = request.ids.correlation_id.as_str(),
224            ocpi.to = request.routing.as_ref().and_then(|r| r.to.as_ref()).map(ToString::to_string),
225            ocpi.from = request.routing.as_ref().map(|r| r.from.to_string()),
226            ocpi.status_code = tracing::field::Empty,
227            http.response.status_code = tracing::field::Empty,
228        );
229        self.attempt_instrumented(request, token, quirks, span.clone()).instrument(span).await
230    }
231
232    async fn attempt_instrumented<T: DeserializeOwned>(
233        &self,
234        request: &OcpiRequest,
235        token: &CredentialsToken,
236        quirks: &Quirks,
237        span: tracing::Span,
238    ) -> Result<(OcpiResponse<T>, HeaderMap), OcpiError> {
239        let mut builder = self
240            .http
241            .request(request.method.clone(), request.url.as_str())
242            .timeout(self.config.timeout)
243            .headers(request.header_map(token, quirks));
244        if let Some(body) = &request.body {
245            builder = builder.body(body.clone());
246        }
247
248        let response = builder.send().await.map_err(|e| OcpiError::Transport(strip_url(&e.to_string())))?;
249        let status = response.status();
250        let headers = response.headers().clone();
251        span.record("http.response.status_code", status.as_u16());
252
253        let bytes = response.bytes().await.map_err(|e| OcpiError::Transport(strip_url(&e.to_string())))?;
254
255        // The five HTTP statuses the specification does use, before the OCPI layer is reached.
256        if !status.is_success() {
257            return Err(match status.as_u16() {
258                400 => OcpiError::MalformedJson(preview(&bytes)),
259                401 => OcpiError::Unauthorized(preview(&bytes)),
260                404 => OcpiError::NotFound(request.url.as_str().to_owned()),
261                405 => OcpiError::MethodNotAllowed(request.url.as_str().to_owned()),
262                other => OcpiError::Transport(format!("HTTP {other}: {}", preview(&bytes))),
263            });
264        }
265
266        let mut de = serde_json::Deserializer::from_slice(&bytes);
267        let envelope: OcpiResponse<T> = serde_path_to_error::deserialize(&mut de).map_err(|e| {
268            OcpiError::Decode { path: e.path().to_string(), message: e.into_inner().to_string() }
269        })?;
270        span.record("ocpi.status_code", envelope.status_code.get());
271        Ok((envelope, headers))
272    }
273}
274
275/// Validates an object before it goes on the wire, when the configuration asks for it.
276///
277/// # Errors
278///
279/// Returns [`OcpiError::Invalid`] listing every violation.
280pub fn check_outgoing<T: Validate>(value: &T, config: &ClientConfig) -> Result<(), OcpiError> {
281    if !config.validate_outgoing {
282        return Ok(());
283    }
284    value.validate().map_err(OcpiError::Invalid)
285}
286
287/// A `reqwest` error message can contain the full URL, including a `response_url` that carries a
288/// one-time token. Keep the cause, drop the URL.
289fn strip_url(message: &str) -> String {
290    match message.find(" for url (") {
291        Some(at) => message[..at].to_owned(),
292        None => message.to_owned(),
293    }
294}
295
296fn preview(bytes: &[u8]) -> String {
297    let text = String::from_utf8_lossy(bytes);
298    let trimmed = text.trim();
299    if trimmed.chars().count() <= 200 {
300        return trimmed.to_owned();
301    }
302    format!("{}…", trimmed.chars().take(200).collect::<String>())
303}
304
305impl RetryPolicy {
306    /// How long to wait before attempt `attempt` (1-based, so the first retry is attempt 2),
307    /// for the request identified by `seed`.
308    ///
309    /// The delay is exponential — `initial_delay * 2^(attempt-1)`, capped at `max_delay` — with
310    /// **equal jitter**: the value actually returned is drawn from the upper half of that
311    /// interval, `[base/2, base]`.
312    ///
313    /// `seed` is what makes the jitter useful. A schedule computed from the attempt number alone
314    /// is identical on every client in a fleet, so a peer that has just come back from an outage
315    /// is hit by all of them at the same instant — the thundering herd that jitter exists to
316    /// prevent. Pass something that differs per client and per request;
317    /// [`RetryPolicy::seed_from`] derives one from the request's `X-Request-ID`, which is a
318    /// freshly generated UUID and therefore already carries the entropy needed.
319    #[must_use]
320    pub fn delay_for(&self, attempt: u32, seed: u64) -> std::time::Duration {
321        let exponent = attempt.saturating_sub(1).min(16);
322        let factor = 1u32 << exponent;
323        let base = self.initial_delay.saturating_mul(factor).min(self.max_delay);
324        let half = base / 2;
325        // A SplitMix64 finaliser over (seed, attempt): a good avalanche in a handful of
326        // instructions, so two clients whose request ids differ in one bit wait very different
327        // amounts, and no random number generator has to be threaded through the client.
328        let spread = half
329            .as_nanos()
330            .try_into()
331            .map_or(0, |span: u64| if span == 0 { 0 } else { mix64(seed ^ u64::from(attempt)) % span });
332        half.saturating_add(std::time::Duration::from_nanos(spread)).min(self.max_delay)
333    }
334
335    /// A [`delay_for`](Self::delay_for) seed derived from a request id.
336    #[must_use]
337    pub fn seed_from(request_id: &str) -> u64 {
338        // FNV-1a: no dependency, and every byte of the UUID reaches the result.
339        let mut hash = 0xcbf2_9ce4_8422_2325_u64;
340        for byte in request_id.as_bytes() {
341            hash ^= u64::from(*byte);
342            hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
343        }
344        hash
345    }
346}
347
348/// The SplitMix64 finalising mix, used to turn a seed into well-distributed jitter.
349const fn mix64(mut z: u64) -> u64 {
350    z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
351    z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
352    z ^ (z >> 31)
353}
354
355#[cfg(test)]
356mod tests {
357    use super::*;
358
359    #[test]
360    fn only_get_may_be_retried() {
361        let url = Url::new("https://e.com/a").unwrap();
362        let get = OcpiRequest::new(Method::GET, url.clone(), ModuleId::Cdrs);
363        let put = OcpiRequest::new(Method::PUT, url.clone(), ModuleId::Cdrs);
364        let post = OcpiRequest::new(Method::POST, url, ModuleId::Cdrs);
365        assert!(get.is_retryable());
366        assert!(!put.is_retryable(), "the spec forbids queueing and retrying writes");
367        assert!(!post.is_retryable());
368    }
369
370    #[test]
371    fn routing_headers_are_dropped_for_configuration_modules() {
372        let url = Url::new("https://e.com/a").unwrap();
373        let routing = RoutingHeaders::new(
374            crate::types::PartyRef::new("NL", "TNM").unwrap(),
375            crate::types::PartyRef::new("DE", "ABC").unwrap(),
376        );
377        let functional =
378            OcpiRequest::new(Method::GET, url.clone(), ModuleId::Locations).routed(routing.clone());
379        assert!(functional.routing.is_some());
380        let configuration = OcpiRequest::new(Method::GET, url, ModuleId::Credentials).routed(routing);
381        assert!(configuration.routing.is_none());
382    }
383
384    #[test]
385    fn the_authorization_header_follows_the_peers_quirks() {
386        let token = CredentialsToken::new("example-token").unwrap();
387        let request = OcpiRequest::new(Method::GET, Url::new("https://e.com/a").unwrap(), ModuleId::Cdrs);
388
389        let modern = request.header_map(&token, &Quirks::default());
390        assert_eq!(modern.get(AUTHORIZATION).unwrap(), "Token ZXhhbXBsZS10b2tlbg==");
391
392        let legacy = request.header_map(&token, &Quirks::for_version(&crate::VersionNumber::V2_1_1));
393        assert_eq!(legacy.get(AUTHORIZATION).unwrap(), "Token example-token");
394    }
395
396    #[test]
397    fn a_body_sets_the_content_type_and_nothing_else_does() {
398        let url = Url::new("https://e.com/a").unwrap();
399        let empty = OcpiRequest::new(Method::GET, url.clone(), ModuleId::Cdrs)
400            .header_map(&CredentialsToken::new("t").unwrap(), &Quirks::default());
401        assert!(empty.get(http::header::CONTENT_TYPE).is_none());
402
403        let with_body = OcpiRequest::new(Method::PUT, url, ModuleId::Cdrs)
404            .with_body(&serde_json::json!({"a": 1}))
405            .unwrap()
406            .header_map(&CredentialsToken::new("t").unwrap(), &Quirks::default());
407        assert_eq!(with_body.get(http::header::CONTENT_TYPE).unwrap(), "application/json");
408    }
409
410    #[test]
411    fn retry_delays_grow_and_stay_under_the_cap() {
412        let policy = RetryPolicy::default();
413        let seed = RetryPolicy::seed_from("6d2b1b3a-0f8f-4e7e-9d3f-1a2b3c4d5e6f");
414        assert!(policy.delay_for(2, seed) > policy.delay_for(1, seed));
415        assert!(policy.delay_for(20, seed) <= policy.max_delay);
416        // Equal jitter: never below half the exponential base, never above it.
417        for attempt in 1..8 {
418            let base = policy.initial_delay.saturating_mul(1 << (attempt - 1)).min(policy.max_delay);
419            let delay = policy.delay_for(attempt, seed);
420            assert!(
421                delay >= base / 2 && delay <= base,
422                "attempt {attempt}: {delay:?} not in half of {base:?}"
423            );
424        }
425    }
426
427    #[test]
428    fn two_clients_retrying_the_same_endpoint_do_not_wait_the_same_time() {
429        // The whole point of jitter: a schedule computed from the attempt number alone would be
430        // identical everywhere, and a peer coming back from an outage would be hit by the whole
431        // fleet at once.
432        let policy = RetryPolicy::default();
433        let delays: std::collections::HashSet<_> =
434            (0..64).map(|i| policy.delay_for(2, RetryPolicy::seed_from(&format!("request-{i}")))).collect();
435        assert!(delays.len() > 50, "only {} distinct delays across 64 requests", delays.len());
436    }
437
438    #[test]
439    fn a_policy_that_does_not_retry_waits_for_nothing() {
440        let policy = RetryPolicy::none();
441        assert_eq!(policy.delay_for(3, 12345), std::time::Duration::ZERO);
442    }
443
444    #[test]
445    fn transport_error_messages_do_not_leak_the_url() {
446        let message = "error sending request for url (https://e.com/cb?token=secret)";
447        assert_eq!(strip_url(message), "error sending request");
448    }
449}
ocpi_kit/client/http.rs

ocpi_kit/client/
http.rs
