aven_core/sync/client/
exchange.rs

1//! Sans-IO HTTP exchanges between the client engine and its host.
2//!
3//! A [`Session`] runs one engine operation. The host repeatedly calls
4//! [`Session::next`]: it sends each [`Step::Request`] exactly as prepared and
5//! answers it with [`Session::accept_response`] or
6//! [`Session::register_transport_failure`], waits out each [`Step::Wait`],
7//! and stops at [`Step::Done`] or an error. Retry and polling decisions stay
8//! in the engine, which expresses them as waits and repeated requests.
9use std::fmt;
10use std::future::{Future, poll_fn};
11use std::pin::Pin;
12use std::sync::{Arc, Mutex, MutexGuard};
13use std::task::Poll;
14use std::time::Duration;
15
16use anyhow::{Result, bail};
17
18/// Timeout for one request, from sending to the last response byte.
19pub(crate) const REQUEST_TIMEOUT: Duration = Duration::from_secs(35);
20/// Slowest upload rate a request body is given time for, beyond
21/// [`REQUEST_TIMEOUT`].
22pub(crate) const MIN_UPLOAD_BYTES_PER_SECOND: u64 = 256 * 1024;
23
24/// [`REQUEST_TIMEOUT`] plus time to send `body_len` bytes at
25/// [`MIN_UPLOAD_BYTES_PER_SECOND`].
26pub(crate) fn request_timeout(body_len: usize) -> Duration {
27    REQUEST_TIMEOUT + Duration::from_millis(body_len as u64 * 1000 / MIN_UPLOAD_BYTES_PER_SECOND)
28}
29
30/// One HTTP header.
31#[derive(Clone, PartialEq, Eq)]
32pub struct HttpHeader {
33    pub name: String,
34    pub value: String,
35}
36
37impl fmt::Debug for HttpHeader {
38    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
39        if self.name.eq_ignore_ascii_case("authorization") {
40            write!(f, "{}: [REDACTED]", self.name)
41        } else {
42            write!(f, "{}: {}", self.name, self.value)
43        }
44    }
45}
46
47/// Names the request a response or failure answers.
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
49pub struct RequestContext {
50    id: u64,
51}
52
53/// A request the host sends exactly as prepared.
54///
55/// The host follows no redirects, uses no proxy or cookies, adds no
56/// `Accept-Encoding` and never decodes content. It reads at most
57/// `response_limit + 1` body bytes and passes them on; the engine refuses
58/// oversized bodies itself. A connection, timeout or body read failure is a
59/// transport failure. The `Authorization` header is a credential.
60pub struct PreparedRequest {
61    pub method: String,
62    pub url: String,
63    pub headers: Vec<HttpHeader>,
64    pub body: Vec<u8>,
65    pub timeout: Duration,
66    pub response_limit: usize,
67    pub context: RequestContext,
68}
69
70impl fmt::Debug for PreparedRequest {
71    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
72        f.debug_struct("PreparedRequest")
73            .field("method", &self.method)
74            .field("url", &self.url)
75            .field("headers", &self.headers)
76            .field("body_len", &self.body.len())
77            .field("timeout", &self.timeout)
78            .field("response_limit", &self.response_limit)
79            .field("context", &self.context)
80            .finish()
81    }
82}
83
84/// A complete HTTP response of any status.
85#[derive(Clone, Debug, PartialEq, Eq)]
86pub struct HttpResponse {
87    pub status: u16,
88    pub headers: Vec<HttpHeader>,
89    pub body: Vec<u8>,
90}
91
92impl HttpResponse {
93    /// The first header named `name`, ignoring case.
94    pub(crate) fn header(&self, name: &str) -> Option<&str> {
95        self.headers
96            .iter()
97            .find(|header| header.name.eq_ignore_ascii_case(name))
98            .map(|header| header.value.as_str())
99    }
100
101    pub(crate) fn has_header(&self, name: &str) -> bool {
102        self.header(name).is_some()
103    }
104
105    pub(crate) fn is_json(&self) -> bool {
106        self.header("content-type") == Some("application/json")
107    }
108
109    pub(crate) fn content_length(&self) -> Option<u64> {
110        self.header("content-length")?.parse().ok()
111    }
112
113    pub(crate) fn retry_after(&self) -> Option<u64> {
114        self.header("retry-after")?.parse().ok()
115    }
116}
117
118/// What the host does next.
119#[derive(Debug)]
120pub enum Step<T> {
121    /// Send this request, then answer it before calling `next` again.
122    Request(PreparedRequest),
123    /// Wait this long, then call `next` again.
124    Wait(Duration),
125    /// The operation finished.
126    Done(T),
127}
128
129/// The request never produced a complete response.
130#[derive(Debug)]
131pub(crate) struct TransportFailure {
132    secure: bool,
133}
134
135enum Outgoing {
136    Request(PreparedRequest),
137    Wait(Duration),
138}
139
140#[derive(Default)]
141enum Pending {
142    #[default]
143    None,
144    Request(u64),
145    Answered(u64, Result<HttpResponse, TransportFailure>),
146    Waiting,
147    Waited,
148}
149
150#[derive(Default)]
151struct LinkState {
152    next_id: u64,
153    outgoing: Option<Outgoing>,
154    pending: Pending,
155}
156
157/// The engine's connection to its host within one session. It carries one
158/// request or wait at a time: callers await each `send` or `wait` before
159/// starting the next, never running two concurrently.
160#[derive(Clone, Default)]
161pub struct Link {
162    state: Arc<Mutex<LinkState>>,
163}
164
165impl Link {
166    fn state(&self) -> MutexGuard<'_, LinkState> {
167        self.state
168            .lock()
169            .unwrap_or_else(|poison| poison.into_inner())
170    }
171
172    /// A state with nothing in flight, ready for the next request or wait.
173    fn idle(&self) -> MutexGuard<'_, LinkState> {
174        let state = self.state();
175        debug_assert!(
176            state.outgoing.is_none() && matches!(state.pending, Pending::None),
177            "a link carries one request or wait at a time"
178        );
179        state
180    }
181
182    /// Hands `request` to the host and waits for its answer.
183    pub(crate) async fn send(
184        &self,
185        method: &str,
186        url: &url::Url,
187        headers: Vec<HttpHeader>,
188        body: Vec<u8>,
189        response_limit: usize,
190    ) -> Result<HttpResponse, TransportFailure> {
191        let id = {
192            let mut state = self.idle();
193            let id = state.next_id;
194            state.next_id += 1;
195            state.outgoing = Some(Outgoing::Request(PreparedRequest {
196                method: method.to_string(),
197                url: url.to_string(),
198                headers,
199                timeout: request_timeout(body.len()),
200                body,
201                response_limit,
202                context: RequestContext { id },
203            }));
204            state.pending = Pending::Request(id);
205            id
206        };
207        poll_fn(|_| {
208            let mut state = self.state();
209            match std::mem::take(&mut state.pending) {
210                Pending::Answered(answered, result) if answered == id => Poll::Ready(result),
211                other => {
212                    state.pending = other;
213                    Poll::Pending
214                }
215            }
216        })
217        .await
218    }
219
220    /// Lets the host wait `delay` before the engine continues.
221    pub(crate) async fn wait(&self, delay: Duration) {
222        {
223            let mut state = self.idle();
224            state.outgoing = Some(Outgoing::Wait(delay));
225            state.pending = Pending::Waiting;
226        }
227        poll_fn(|_| {
228            let mut state = self.state();
229            if matches!(state.pending, Pending::Waited) {
230                state.pending = Pending::None;
231                Poll::Ready(())
232            } else {
233                Poll::Pending
234            }
235        })
236        .await
237    }
238
239    fn answer(
240        &self,
241        context: RequestContext,
242        result: Result<HttpResponse, TransportFailure>,
243    ) -> Result<()> {
244        let mut state = self.state();
245        match state.pending {
246            Pending::Request(id) if id == context.id && state.outgoing.is_none() => {
247                state.pending = Pending::Answered(id, result);
248                Ok(())
249            }
250            _ => bail!("error sync-exchange-unexpected-answer"),
251        }
252    }
253}
254
255type Operation<'a, T> = Pin<Box<dyn Future<Output = Result<T>> + Send + 'a>>;
256
257/// One engine operation driven by its host.
258pub struct Session<'a, T> {
259    operation: Option<Operation<'a, T>>,
260    link: Link,
261}
262
263impl<'a, T> Session<'a, T> {
264    /// Runs `operation` with a link to this session's host.
265    pub fn new<F, Fut>(operation: F) -> Self
266    where
267        F: FnOnce(Link) -> Fut,
268        Fut: Future<Output = Result<T>> + Send + 'a,
269    {
270        let link = Link::default();
271        Self {
272            operation: Some(Box::pin(operation(link.clone()))),
273            link,
274        }
275    }
276
277    /// Runs the operation until it needs the host or finishes. An error
278    /// finishes the session.
279    pub async fn next(&mut self) -> Result<Step<T>> {
280        {
281            let mut state = self.link.state();
282            match state.pending {
283                Pending::Request(_) if state.outgoing.is_none() => {
284                    bail!("error sync-exchange-unanswered")
285                }
286                Pending::Waiting if state.outgoing.is_none() => state.pending = Pending::Waited,
287                _ => {}
288            }
289        }
290        let Some(operation) = self.operation.as_mut() else {
291            bail!("error sync-session-finished");
292        };
293        let link = &self.link;
294        let finished = poll_fn(|context| {
295            if let Poll::Ready(result) = operation.as_mut().poll(context) {
296                return Poll::Ready(Some(result));
297            }
298            if link.state().outgoing.is_some() {
299                return Poll::Ready(None);
300            }
301            Poll::Pending
302        })
303        .await;
304        if let Some(result) = finished {
305            self.operation = None;
306            return result.map(Step::Done);
307        }
308        Ok(match self.link.state().outgoing.take() {
309            Some(Outgoing::Request(request)) => Step::Request(request),
310            Some(Outgoing::Wait(delay)) => Step::Wait(delay),
311            None => unreachable!("outgoing step observed above"),
312        })
313    }
314
315    /// Answers the outstanding request with the response the host received.
316    pub fn accept_response(
317        &mut self,
318        context: RequestContext,
319        response: HttpResponse,
320    ) -> Result<()> {
321        self.link.answer(context, Ok(response))
322    }
323
324    /// Answers the outstanding request that produced no complete response.
325    pub fn register_transport_failure(&mut self, context: RequestContext) -> Result<()> {
326        self.link
327            .answer(context, Err(TransportFailure { secure: false }))
328    }
329
330    /// Answers an HTTPS request that could not establish or retain its secure
331    /// connection. Hosts need not expose certificate or endpoint details.
332    pub fn register_secure_transport_failure(&mut self, context: RequestContext) -> Result<()> {
333        self.link
334            .answer(context, Err(TransportFailure { secure: true }))
335    }
336}
337
338/// Busy retries after the first attempt of one JSON exchange.
339const BUSY_RETRIES: usize = 3;
340/// Refusal bodies over this many bytes carry no readable error code.
341const REFUSAL_LIMIT: usize = 1024;
342
343/// Why a JSON exchange produced no reply.
344#[derive(Debug)]
345pub(crate) enum Failure {
346    /// No complete response arrived over HTTP.
347    Network,
348    /// No complete response arrived over HTTPS.
349    SecureTransport,
350    /// An unstructured 413 response, distinct from server limit codes.
351    RequestBodyLimit,
352    /// A non-success status with the `{"error":"<code>"}` body's code, if
353    /// readable. A busy server was already retried.
354    Refused { status: u16, code: Option<String> },
355    /// A success response that is not uncompressed JSON.
356    Malformed,
357    /// A success response over its limit.
358    TooLarge,
359}
360
361/// POSTs a JSON `body` and returns the success body of at most
362/// `response_limit` bytes. A busy server, answering 503 with `Retry-After`,
363/// is retried a bounded number of times.
364pub(crate) async fn post_json(
365    link: &Link,
366    endpoint: &url::Url,
367    credential: Option<&crate::sync::seed_claim::Secret>,
368    body: Vec<u8>,
369    response_limit: usize,
370) -> Result<Vec<u8>, Failure> {
371    post(
372        link,
373        endpoint,
374        credential,
375        json_content(),
376        body,
377        response_limit,
378    )
379    .await
380}
381
382/// [`post_json`] with a body of any content type and a JSON reply.
383pub(crate) async fn post(
384    link: &Link,
385    endpoint: &url::Url,
386    credential: Option<&crate::sync::seed_claim::Secret>,
387    content_type: HttpHeader,
388    body: Vec<u8>,
389    response_limit: usize,
390) -> Result<Vec<u8>, Failure> {
391    let mut headers = vec![content_type];
392    headers.extend(credential.map(bearer));
393    let mut attempt = 0;
394    loop {
395        let response = link
396            .send(
397                "POST",
398                endpoint,
399                headers.clone(),
400                body.clone(),
401                response_limit,
402            )
403            .await
404            .map_err(|failure| {
405                if failure.secure {
406                    Failure::SecureTransport
407                } else {
408                    Failure::Network
409                }
410            })?;
411        if response.status == 200 {
412            if !response.is_json() || response.has_header("content-encoding") {
413                return Err(Failure::Malformed);
414            }
415            if response.body.len() > response_limit
416                || response
417                    .content_length()
418                    .is_some_and(|length| length > response_limit as u64)
419            {
420                return Err(Failure::TooLarge);
421            }
422            return Ok(response.body);
423        }
424        let expected_busy = match endpoint.path() {
425            "/e2ee/tail/v1" | "/e2ee/tail/batch/v1" => Some("encrypted-tail-busy"),
426            "/e2ee/images/v1" => Some("encrypted-image-busy"),
427            "/e2ee/enrollment/v1" => Some("enrollment-busy"),
428            "/e2ee/bootstrap/v1" => Some("bootstrap-busy"),
429            _ => None,
430        };
431        if response.status == 503
432            && expected_busy.is_some()
433            && refusal_code(&response).as_deref() == expected_busy
434            && attempt < BUSY_RETRIES
435            && let Some(retry_after) = response.retry_after()
436        {
437            link.wait(busy_retry_delay(attempt, retry_after)).await;
438            attempt += 1;
439            continue;
440        }
441        let code = refusal_code(&response);
442        if response.status == 413 && code.is_none() {
443            return Err(Failure::RequestBodyLimit);
444        }
445        return Err(Failure::Refused {
446            status: response.status,
447            code,
448        });
449    }
450}
451
452fn refusal_code(response: &HttpResponse) -> Option<String> {
453    #[derive(serde::Deserialize)]
454    struct Refusal {
455        error: String,
456    }
457    if !response.is_json() || response.body.len() > REFUSAL_LIMIT {
458        return None;
459    }
460    serde_json::from_slice::<Refusal>(&response.body)
461        .ok()
462        .map(|refusal| refusal.error)
463}
464
465/// Backoff before retrying a busy server; never shorter than its
466/// `Retry-After`, capped at two seconds.
467fn busy_retry_delay(attempt: usize, retry_after: u64) -> Duration {
468    let base_ms = 50_u64 << attempt.min(5);
469    let mut random = [0_u8; 2];
470    let _ = getrandom::fill(&mut random);
471    let jitter_ms = u16::from_le_bytes(random) as u64 % (base_ms / 2 + 1);
472    Duration::from_millis(base_ms + jitter_ms).max(Duration::from_secs(retry_after.min(2)))
473}
474
475/// A bearer credential header.
476pub(crate) fn bearer(secret: &crate::sync::seed_claim::Secret) -> HttpHeader {
477    HttpHeader {
478        name: "authorization".into(),
479        value: format!("Bearer {}", hex::encode(secret.expose())),
480    }
481}
482
483pub(crate) fn json_content() -> HttpHeader {
484    HttpHeader {
485        name: "content-type".into(),
486        value: "application/json".into(),
487    }
488}
489
490#[cfg(test)]
491mod tests {
492    use super::*;
493
494    async fn two_requests(link: Link) -> Result<Vec<u16>> {
495        let url = url::Url::parse("https://sync.example.com/x").unwrap();
496        let mut statuses = Vec::new();
497        for _ in 0..2 {
498            match link
499                .send("POST", &url, vec![json_content()], vec![1], 8)
500                .await
501            {
502                Ok(response) => statuses.push(response.status),
503                Err(TransportFailure { .. }) => statuses.push(0),
504            }
505            link.wait(Duration::from_millis(5)).await;
506        }
507        Ok(statuses)
508    }
509
510    #[tokio::test]
511    async fn session_yields_requests_and_waits_in_order() {
512        let mut session = Session::new(two_requests);
513        let Step::Request(first) = session.next().await.unwrap() else {
514            panic!("expected a request");
515        };
516        assert_eq!(first.method, "POST");
517        assert_eq!(first.timeout, REQUEST_TIMEOUT);
518        assert!(session.next().await.is_err(), "an unanswered request stops");
519        let response = HttpResponse {
520            status: 204,
521            headers: vec![],
522            body: vec![],
523        };
524        session.accept_response(first.context, response).unwrap();
525        assert!(matches!(
526            session.next().await.unwrap(),
527            Step::Wait(delay) if delay == Duration::from_millis(5)
528        ));
529        let Step::Request(second) = session.next().await.unwrap() else {
530            panic!("expected a request");
531        };
532        assert!(
533            session
534                .accept_response(
535                    first.context,
536                    HttpResponse {
537                        status: 200,
538                        headers: vec![],
539                        body: vec![]
540                    }
541                )
542                .is_err(),
543            "a stale context is refused"
544        );
545        session.register_transport_failure(second.context).unwrap();
546        assert!(matches!(session.next().await.unwrap(), Step::Wait(_)));
547        assert!(
548            matches!(session.next().await.unwrap(), Step::Done(statuses) if statuses == [204, 0])
549        );
550        assert!(session.next().await.is_err());
551    }
552
553    #[tokio::test]
554    async fn secure_transport_failures_remain_distinct() {
555        let url = url::Url::parse("https://sync.example.com/x").unwrap();
556        let mut session = Session::new(|link| async move {
557            Ok(matches!(
558                post_json(&link, &url, None, vec![], 8).await,
559                Err(Failure::SecureTransport)
560            ))
561        });
562        let Step::Request(request) = session.next().await.unwrap() else {
563            panic!("expected a request");
564        };
565        session
566            .register_secure_transport_failure(request.context)
567            .unwrap();
568        assert!(matches!(session.next().await.unwrap(), Step::Done(true)));
569    }
570
571    #[tokio::test]
572    async fn unstructured_413_is_a_request_body_limit() {
573        let url = url::Url::parse("https://sync.example.com/x").unwrap();
574        let mut session = Session::new(|link| async move {
575            Ok(matches!(
576                post_json(&link, &url, None, vec![], 8).await,
577                Err(Failure::RequestBodyLimit)
578            ))
579        });
580        let Step::Request(request) = session.next().await.unwrap() else {
581            panic!("expected a request");
582        };
583        session
584            .accept_response(
585                request.context,
586                HttpResponse {
587                    status: 413,
588                    headers: vec![],
589                    body: b"request too large".to_vec(),
590                },
591            )
592            .unwrap();
593        assert!(matches!(session.next().await.unwrap(), Step::Done(true)));
594    }
595
596    #[test]
597    fn credentials_stay_out_of_debug_output() {
598        let secret = crate::sync::seed_claim::Secret::new([7; 32]);
599        let header = bearer(&secret);
600        assert!(!format!("{header:?}").contains(&hex::encode([7; 32])));
601    }
602}
aven_core/sync/client/exchange.rs

aven_core/sync/client/
exchange.rs
