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    /// An explicit endpoint-family hosting refusal, never device authority.
356    Hosting(aven_protocol::refusal::Hosting),
357    /// A success response that is not uncompressed JSON.
358    Malformed,
359    /// A success response over its limit.
360    TooLarge,
361}
362
363/// POSTs a JSON `body` and returns the success body of at most
364/// `response_limit` bytes. A busy server, answering 503 with `Retry-After`,
365/// is retried a bounded number of times.
366pub(crate) async fn post_json(
367    link: &Link,
368    endpoint: &url::Url,
369    credential: Option<&crate::sync::seed_claim::Secret>,
370    body: Vec<u8>,
371    response_limit: usize,
372) -> Result<Vec<u8>, Failure> {
373    post(
374        link,
375        endpoint,
376        credential,
377        json_content(),
378        body,
379        response_limit,
380    )
381    .await
382}
383
384/// [`post_json`] with a body of any content type and a JSON reply.
385pub(crate) async fn post(
386    link: &Link,
387    endpoint: &url::Url,
388    credential: Option<&crate::sync::seed_claim::Secret>,
389    content_type: HttpHeader,
390    body: Vec<u8>,
391    response_limit: usize,
392) -> Result<Vec<u8>, Failure> {
393    let mut headers = vec![content_type];
394    headers.extend(credential.map(bearer));
395    let mut attempt = 0;
396    loop {
397        let response = link
398            .send(
399                "POST",
400                endpoint,
401                headers.clone(),
402                body.clone(),
403                response_limit,
404            )
405            .await
406            .map_err(|failure| {
407                if failure.secure {
408                    Failure::SecureTransport
409                } else {
410                    Failure::Network
411                }
412            })?;
413        if response.status == 200 {
414            if !response.is_json() || response.has_header("content-encoding") {
415                return Err(Failure::Malformed);
416            }
417            if response.body.len() > response_limit
418                || response
419                    .content_length()
420                    .is_some_and(|length| length > response_limit as u64)
421            {
422                return Err(Failure::TooLarge);
423            }
424            return Ok(response.body);
425        }
426        let expected_busy = match endpoint.path() {
427            "/e2ee/tail/v1" | "/e2ee/tail/batch/v1" => Some("encrypted-tail-busy"),
428            "/e2ee/images/v1" => Some("encrypted-image-busy"),
429            "/e2ee/enrollment/v1" => Some("enrollment-busy"),
430            "/e2ee/bootstrap/v1" => Some("bootstrap-busy"),
431            _ => None,
432        };
433        if response.status == 503
434            && expected_busy.is_some()
435            && refusal_code(&response).as_deref() == expected_busy
436            && attempt < BUSY_RETRIES
437            && let Some(retry_after) = response.retry_after()
438        {
439            link.wait(busy_retry_delay(attempt, retry_after)).await;
440            attempt += 1;
441            continue;
442        }
443        let code = refusal_code(&response);
444        use aven_protocol::refusal::{
445            BOOTSTRAP_HOSTING, ENROLLMENT_HOSTING, IMAGE_HOSTING, TAIL_HOSTING,
446        };
447        let hosting = match endpoint.path() {
448            "/e2ee/bootstrap/v1" => Some(&BOOTSTRAP_HOSTING),
449            "/e2ee/enrollment/v1" => Some(&ENROLLMENT_HOSTING),
450            "/e2ee/tail/v1" | "/e2ee/tail/batch/v1" => Some(&TAIL_HOSTING),
451            "/e2ee/images/v1" => Some(&IMAGE_HOSTING),
452            _ => None,
453        };
454        if let Some(hosting) =
455            hosting.and_then(|codes| codes.classify(response.status, code.as_deref()))
456        {
457            return Err(Failure::Hosting(hosting));
458        }
459        if response.status == 413 && code.is_none() {
460            return Err(Failure::RequestBodyLimit);
461        }
462        return Err(Failure::Refused {
463            status: response.status,
464            code,
465        });
466    }
467}
468
469fn refusal_code(response: &HttpResponse) -> Option<String> {
470    #[derive(serde::Deserialize)]
471    struct Refusal {
472        error: String,
473    }
474    if !response.is_json() || response.body.len() > REFUSAL_LIMIT {
475        return None;
476    }
477    serde_json::from_slice::<Refusal>(&response.body)
478        .ok()
479        .map(|refusal| refusal.error)
480}
481
482/// Backoff before retrying a busy server; never shorter than its
483/// `Retry-After`, capped at two seconds.
484fn busy_retry_delay(attempt: usize, retry_after: u64) -> Duration {
485    let base_ms = 50_u64 << attempt.min(5);
486    let mut random = [0_u8; 2];
487    let _ = getrandom::fill(&mut random);
488    let jitter_ms = u16::from_le_bytes(random) as u64 % (base_ms / 2 + 1);
489    Duration::from_millis(base_ms + jitter_ms).max(Duration::from_secs(retry_after.min(2)))
490}
491
492/// A bearer credential header.
493pub(crate) fn bearer(secret: &crate::sync::seed_claim::Secret) -> HttpHeader {
494    HttpHeader {
495        name: "authorization".into(),
496        value: format!("Bearer {}", hex::encode(secret.expose())),
497    }
498}
499
500pub(crate) fn json_content() -> HttpHeader {
501    HttpHeader {
502        name: "content-type".into(),
503        value: "application/json".into(),
504    }
505}
506
507#[cfg(test)]
508mod tests {
509    use super::*;
510
511    async fn two_requests(link: Link) -> Result<Vec<u16>> {
512        let url = url::Url::parse("https://sync.example.com/x").unwrap();
513        let mut statuses = Vec::new();
514        for _ in 0..2 {
515            match link
516                .send("POST", &url, vec![json_content()], vec![1], 8)
517                .await
518            {
519                Ok(response) => statuses.push(response.status),
520                Err(TransportFailure { .. }) => statuses.push(0),
521            }
522            link.wait(Duration::from_millis(5)).await;
523        }
524        Ok(statuses)
525    }
526
527    #[tokio::test]
528    async fn session_yields_requests_and_waits_in_order() {
529        let mut session = Session::new(two_requests);
530        let Step::Request(first) = session.next().await.unwrap() else {
531            panic!("expected a request");
532        };
533        assert_eq!(first.method, "POST");
534        assert_eq!(first.timeout, REQUEST_TIMEOUT);
535        assert!(session.next().await.is_err(), "an unanswered request stops");
536        let response = HttpResponse {
537            status: 204,
538            headers: vec![],
539            body: vec![],
540        };
541        session.accept_response(first.context, response).unwrap();
542        assert!(matches!(
543            session.next().await.unwrap(),
544            Step::Wait(delay) if delay == Duration::from_millis(5)
545        ));
546        let Step::Request(second) = session.next().await.unwrap() else {
547            panic!("expected a request");
548        };
549        assert!(
550            session
551                .accept_response(
552                    first.context,
553                    HttpResponse {
554                        status: 200,
555                        headers: vec![],
556                        body: vec![]
557                    }
558                )
559                .is_err(),
560            "a stale context is refused"
561        );
562        session.register_transport_failure(second.context).unwrap();
563        assert!(matches!(session.next().await.unwrap(), Step::Wait(_)));
564        assert!(
565            matches!(session.next().await.unwrap(), Step::Done(statuses) if statuses == [204, 0])
566        );
567        assert!(session.next().await.is_err());
568    }
569
570    #[tokio::test]
571    async fn secure_transport_failures_remain_distinct() {
572        let url = url::Url::parse("https://sync.example.com/x").unwrap();
573        let mut session = Session::new(|link| async move {
574            Ok(matches!(
575                post_json(&link, &url, None, vec![], 8).await,
576                Err(Failure::SecureTransport)
577            ))
578        });
579        let Step::Request(request) = session.next().await.unwrap() else {
580            panic!("expected a request");
581        };
582        session
583            .register_secure_transport_failure(request.context)
584            .unwrap();
585        assert!(matches!(session.next().await.unwrap(), Step::Done(true)));
586    }
587
588    #[tokio::test]
589    async fn unstructured_413_is_a_request_body_limit() {
590        let url = url::Url::parse("https://sync.example.com/x").unwrap();
591        let mut session = Session::new(|link| async move {
592            Ok(matches!(
593                post_json(&link, &url, None, vec![], 8).await,
594                Err(Failure::RequestBodyLimit)
595            ))
596        });
597        let Step::Request(request) = session.next().await.unwrap() else {
598            panic!("expected a request");
599        };
600        session
601            .accept_response(
602                request.context,
603                HttpResponse {
604                    status: 413,
605                    headers: vec![],
606                    body: b"request too large".to_vec(),
607                },
608            )
609            .unwrap();
610        assert!(matches!(session.next().await.unwrap(), Step::Done(true)));
611    }
612
613    #[test]
614    fn credentials_stay_out_of_debug_output() {
615        let secret = crate::sync::seed_claim::Secret::new([7; 32]);
616        let header = bearer(&secret);
617        assert!(!format!("{header:?}").contains(&hex::encode([7; 32])));
618    }
619}
aven_core/sync/client/exchange.rs

aven_core/sync/client/
exchange.rs
